#!/usr/bin/env python # -*- coding: utf-8 -*- import requests import json import random import time # API基础URL BASE_URL = "http://127.0.0.1:9999/api/v1" # 测试数据 test_phone = f"1380000{random.randint(1000, 9999)}" test_password = test_phone[-6:] # 默认密码是手机号后6位 access_token = None user_id = None valuation_id = None def test_register(): """测试用户注册功能""" print("\n===== 测试用户注册 =====") url = f"{BASE_URL}/app-user/register" data = { "phone": test_phone } response = requests.post(url, json=data) print(f"请求URL: {url}") print(f"请求数据: {data}") print(f"响应状态码: {response.status_code}") print(f"响应内容: {response.text}") assert response.status_code == 200, "注册请求失败" result = response.json() assert result["code"] == 200, "注册失败" assert result["data"]["phone"] == test_phone, "返回的手机号不匹配" assert result["data"]["default_password"] == test_phone[-6:], "默认密码不正确" print("✅ 用户注册测试通过") return result def test_login(): """测试用户登录功能""" global access_token print("\n===== 测试用户登录 =====") url = f"{BASE_URL}/app-user/login" data = { "phone": test_phone, "password": test_password } response = requests.post(url, json=data) print(f"请求URL: {url}") print(f"请求数据: {data}") print(f"响应状态码: {response.status_code}") print(f"响应内容: {response.text}") assert response.status_code == 200, "登录请求失败" result = response.json() assert "access_token" in result, "登录失败，未返回token" # 保存token供后续请求使用 access_token = result["access_token"] print("✅ 用户登录测试通过") return result def test_create_valuation(): """测试创建估值评估申请""" global access_token print("\n===== 测试创建估值评估申请 =====") url = f"{BASE_URL}/app-valuations/" # 准备请求头，包含授权token headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json" } # 估值评估申请数据 - 根据估值字段.txt更新 data = { # 02 - 基础信息 - 非遗IP资产的基本信息 "asset_name": f"蜀绣-{random.randint(1000, 9999)}", # 资产名称：必须是企业全名称 "institution": "有数", # 所属机构：拥有或管理该非遗IP资产的机构名称 "industry": "文化艺术", # 所属行业：非遗IP资产所属的行业分类 # 03 - 财务状况 "rd_investment": "5", # ：近12个月的研发费用（单位：万元） "annual_revenue": "100", # 近12个月的营收额（单位：万元），用于计算创新投入比 "three_year_income": [100, 120, 150], # 近三年的每年收益：资产近3年的年收益数据（单位：万元），用于计算年均收益和增长率 "funding_status": "国家级资助", # 资金支持: 国家级资助(10分)、省级资助(7分)、无资助(0分) # 04 - 非遗等级与技术 "inheritor_level": "国家级", # 传承人等级 "inheritor_ages": [60, 42, 35], # 传承人年龄 "inheritor_certificates": ["http://example.com/国家级非遗传承人证书.jpg"], # 传承人证书：传承人资质证明材料 "heritage_asset_level": "国家级非遗", # 非遗资产等级 "patent_remaining_years": "8", # [实际上就是专利号 通过API查询到的 ]专利剩余年限：资产相关专利的剩余保护期，8年对应7分 "historical_evidence": { # 资产历史证据类型+数量：历史传承的证据材料 "artifacts": 1, # 出土实物数量 "ancient_literature": 2, # 古代文献数量 "inheritor_testimony": 3, # 传承人佐证数量 "modern_research": 1 # 现代研究数量 }, # 专利证书： "patent_certificates": ["http://example.com/专利证书1.jpg", "http://example.com/专利证书2.jpg"], "pattern_images": ["pattern1.jpg"], # 纹样图片：资产相关的纹样图片文件 # 04 - 非遗应用与推广 "implementation_stage": "成熟应用", # 非遗资产应用成熟度 "coverage_area": "区域覆盖", # 非遗资产应用覆盖范围 "collaboration_type": "品牌联名", # 非遗资产跨界合作深度 "offline_teaching_count": 12, # 近12个月线下相关演讲活动次数 "platform_accounts": { # 线上相关宣传账号信息 "bilibili": { "followers_count": 8000, # 粉丝数量 "likes": 1000, # 点赞数 "comments": 500, # 评论数 "shares": 500 # 转发数 }, # B站账号 "douyin": { "followers_count": 8000, # 粉丝数量 "likes": 1000, # 点赞数 "comments": 500, # 评论数 "shares": 500 # 转发数 } # 抖音账号 }, # 06 - 非遗资产衍生商品信息 #该商品近12个月销售量 "sales_volume": "1000", # 近12个月销售量：资产衍生商品的近12个月销售量（单位：件） 链接购买量 # 该商品近12个月的链接浏览量 "link_views": "10000", # 近12个月链接浏览量：资产衍生商品相关链接的近12个月浏览量（单位：次） 浏览量 "scarcity_level": "流通", # 稀缺等级：资产的稀缺程度，流通（发行量>1000份）对应0.1分 "market_activity_time": "近一月", # 市场活动的时间 "monthly_transaction_amount": "<100万元", # 月交易额：资产衍生商品的月交易额水平，<100万元对应-0.1 "price_range": { # 资产商品的价格波动率：近30天商品价格的波动情况 "highest": 239, # 最高价（单位：元） "lowest": 189 # 最低价（单位：元） }, "market_price": 0, # 直接提供的市场价格（单位：万元） 用户输入: 专家审核 或者 系统默认 专家审核 # 内置API 计算字段 "infringement_record": "无侵权记录", # 侵权记录：资产的侵权历史情况，无侵权记录对应10分 "patent_count": "1", # 专利使用量：资产相关的专利数量，每引用一项专利+2.5分 "esg_value": "10", # ESG关联价值：根据行业匹配的ESG（环境、社会、治理）关联价值 "policy_matching": "10", # 政策匹配度：根据行业自动匹配的政策匹配度分值 "online_course_views": 2000, # 线上课程点击量：抖音/快手播放量按100:1折算为学习人次，B站课程按50:1折算 "pattern_complexity": "1.459", # 结构复杂度：纹样的结构复杂度值 搞一个默认值: 0.0 "normalized_entropy": "9.01", # 归一化信息熵：纹样的归一化信息熵值 搞一个默认值: 0.0 "legal_risk": "无诉讼", # 法律风险-侵权诉讼历史：资产所属机构的诉讼历史，无诉讼对应10分 # 动态质押率DPR：计算公式=基础质押率*(1+流量修正系数)+政策加成系数-流动性调节因子 "base_pledge_rate": "50%", # 基础质押率：基础质押率固定值50% "flow_correction": "0.3", # 流量修正系数：固定值0.3 } response = requests.post(url, headers=headers, json=data) print(f"请求URL: {url}") print(f"请求头: {headers}") print(f"请求数据: {json.dumps(data, ensure_ascii=False, indent=2)}") print(f"响应状态码: {response.status_code}") print(f"响应内容: {response.text}") assert response.status_code == 200, "创建估值评估申请请求失败" result = response.json() assert result["code"] == 200, "创建估值评估申请失败" print("✅ 创建估值评估申请测试通过") return result def test_get_profile(): """测试获取用户个人信息""" global access_token, user_id print("\n===== 测试获取用户个人信息 =====") url = f"{BASE_URL}/app-user/profile" headers = { "Authorization": f"Bearer {access_token}" } response = requests.get(url, headers=headers) print(f"请求URL: {url}") print(f"请求头: {headers}") print(f"响应状态码: {response.status_code}") print(f"响应内容: {response.text}") assert response.status_code == 200, "获取用户信息请求失败" result = response.json() user_id = result["id"] # 保存用户ID供后续使用 print("✅ 获取用户个人信息测试通过") return result def test_change_password(): """测试修改密码""" global access_token print("\n===== 测试修改密码 =====") url = f"{BASE_URL}/app-user/change-password" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json" } new_password = "new" + test_password data = { "old_password": test_password, "new_password": new_password } response = requests.post(url, headers=headers, json=data) print(f"请求URL: {url}") print(f"请求头: {headers}") print(f"请求数据: {data}") print(f"响应状态码: {response.status_code}") print(f"响应内容: {response.text}") assert response.status_code == 200, "修改密码请求失败" result = response.json() assert result["code"] == 200, "修改密码失败" print("✅ 修改密码测试通过") return result def test_update_profile(): """测试更新用户信息""" global access_token print("\n===== 测试更新用户信息 =====") url = f"{BASE_URL}/app-user/profile" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json" } data = { "nickname": f"测试用户{random.randint(100, 999)}", "avatar": "https://example.com/avatar.jpg", "gender": "male", "email": f"test{random.randint(100, 999)}@example.com" } response = requests.put(url, headers=headers, json=data) print(f"请求URL: {url}") print(f"请求头: {headers}") print(f"请求数据: {data}") print(f"响应状态码: {response.status_code}") print(f"响应内容: {response.text}") assert response.status_code == 200, "更新用户信息请求失败" result = response.json() # 更新用户信息接口直接返回用户对象，不包含code字段 assert "id" in result, "更新用户信息失败" print("✅ 更新用户信息测试通过") return result def test_logout(): """测试用户登出""" global access_token print("\n===== 测试用户登出 =====") url = f"{BASE_URL}/app-user/logout" headers = { "Authorization": f"Bearer {access_token}" } response = requests.post(url, headers=headers) print(f"请求URL: {url}") print(f"请求头: {headers}") print(f"响应状态码: {response.status_code}") print(f"响应内容: {response.text}") assert response.status_code == 200, "登出请求失败" result = response.json() assert result["code"] == 200, "登出失败" print("✅ 用户登出测试通过") return result def test_get_valuation_list(): """测试获取用户估值列表""" global access_token print("\n===== 测试获取用户估值列表 =====") url = f"{BASE_URL}/app-valuations/" headers = { "Authorization": f"Bearer {access_token}" } response = requests.get(url, headers=headers) print(f"请求URL: {url}") print(f"请求头: {headers}") print(f"响应状态码: {response.status_code}") print(f"响应内容: {response.text}") assert response.status_code == 200, "获取估值列表请求失败" result = response.json() assert result["code"] == 200, "获取估值列表失败" print("✅ 获取用户估值列表测试通过") return result def test_get_valuation_detail(): """测试获取估值详情""" global access_token, valuation_id # 先获取估值列表，获取第一个估值ID if not valuation_id: list_result = test_get_valuation_list() if list_result["data"] and len(list_result["data"]) > 0: valuation_id = list_result["data"][0]["id"] else: print("⚠️ 没有可用的估值记录，跳过估值详情测试") return None print("\n===== 测试获取估值详情 =====") url = f"{BASE_URL}/app-valuations/{valuation_id}" headers = { "Authorization": f"Bearer {access_token}" } response = requests.get(url, headers=headers) print(f"请求URL: {url}") print(f"请求头: {headers}") print(f"响应状态码: {response.status_code}") print(f"响应内容: {response.text}") assert response.status_code == 200, "获取估值详情请求失败" result = response.json() assert result["code"] == 200, "获取估值详情失败" print("✅ 获取估值详情测试通过") return result def run_tests(): """运行所有测试""" try: # 测试注册 test_register() # 等待一秒，确保数据已保存 time.sleep(1) # 测试登录 test_login() # 测试获取用户个人信息 test_get_profile() # 测试更新用户信息 test_update_profile() # 测试创建估值评估申请 test_create_valuation() # 测试获取估值列表 test_get_valuation_list() # 测试获取估值详情 test_get_valuation_detail() # 测试修改密码 test_change_password() # 测试登出 test_logout() print("\n===== 所有测试通过 =====") except AssertionError as e: print(f"\n❌ 测试失败: {e}") except Exception as e: print(f"\n❌ 发生错误: {e}") if __name__ == "__main__": run_tests()