test_tutorial_02.py

"""
教程 02 代码验证脚本

验证内容：
1. 模块导入正确
2. 数据模型定义正确
3. 工具函数可以独立调用
4. Pydantic 校验工作正常
"""

import sys
import os

# 修复 Windows 终端编码问题
if sys.platform == 'win32':
    import io
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')

# 添加父目录到路径
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

print("=" * 80)
print("教程 02 代码验证")
print("=" * 80)

# ============================================
# 测试 1：基础工具函数（无需 API）
# ============================================
print("\n【测试 1】基础工具函数")
print("-" * 60)

# 模拟导入并测试工具函数
def get_weather(city: str, unit: str = "celsius") -> str:
    """获取天气信息"""
    weather_data = {
        "北京": {"temp": 25, "condition": "晴朗"},
        "上海": {"temp": 28, "condition": "多云"},
    }

    data = weather_data.get(city, {"temp": 20, "condition": "未知"})
    temp = data["temp"]

    if unit == "fahrenheit":
        temp = temp * 9/5 + 32
        unit_symbol = "°F"
    else:
        unit_symbol = "°C"

    return f"{city}的天气：{data['condition']}，温度 {temp}{unit_symbol}"

# 测试
result1 = get_weather("北京")
assert "北京" in result1 and "25" in result1
print(f"✓ get_weather('北京') = {result1}")

result2 = get_weather("上海", "fahrenheit")
assert "上海" in result2 and "82" in result2
print(f"✓ get_weather('上海', 'fahrenheit') = {result2}")

print("✅ 基础工具函数测试通过")


# ============================================
# 测试 2：Pydantic 参数校验
# ============================================
print("\n【测试 2】Pydantic 参数校验")
print("-" * 60)

from pydantic import BaseModel, Field, field_validator, ValidationError

class QueryOrderParams(BaseModel):
    """订单查询参数"""
    order_id: str = Field(
        ...,
        description="订单号",
        pattern=r"^ORD\d{10}$"
    )

    @field_validator("order_id")
    @classmethod
    def validate_order_id(cls, v: str) -> str:
        if "'" in v or '"' in v or ';' in v:
            raise ValueError("订单号包含非法字符")
        return v

# 测试正常情况
try:
    params1 = QueryOrderParams(order_id="ORD1234567890")
    print(f"✓ 正常订单号: {params1.order_id}")
except ValidationError as e:
    print(f"✗ 错误: {e}")

# 测试格式错误
try:
    params2 = QueryOrderParams(order_id="ABC1234567890")
    print(f"✗ 应该抛出错误但没有: {params2.order_id}")
except ValidationError as e:
    print(f"✓ 正确拦截格式错误: {e.error_count()} 个错误")

# 测试注入攻击
try:
    params3 = QueryOrderParams(order_id="ORD1234567890'; DROP TABLE orders;--")
    print(f"✗ 应该拦截 SQL 注入")
except ValidationError as e:
    print(f"✓ 正确拦截 SQL 注入: {e.error_count()} 个错误")

print("✅ Pydantic 参数校验测试通过")


# ============================================
# 测试 3：风险等级枚举
# ============================================
print("\n【测试 3】风险等级枚举")
print("-" * 60)

from enum import Enum

class RiskLevel(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

print(f"✓ RiskLevel.LOW = {RiskLevel.LOW}")
print(f"✓ RiskLevel.MEDIUM = {RiskLevel.MEDIUM}")
print(f"✓ RiskLevel.HIGH = {RiskLevel.HIGH}")

assert RiskLevel.LOW.value == "low"
assert RiskLevel.MEDIUM.value == "medium"
assert RiskLevel.HIGH.value == "high"

print("✅ 风险等级枚举测试通过")


# ============================================
# 测试 4：客服 Agent 数据模型
# ============================================
print("\n【测试 4】客服 Agent 数据模型")
print("-" * 60)

from typing import List
from datetime import datetime

class Order:
    """订单模型"""
    def __init__(self, order_id: str, user_id: str, status: str,
                 amount: float, items: List[str], create_time: str):
        self.order_id = order_id
        self.user_id = user_id
        self.status = status
        self.amount = amount
        self.items = items
        self.create_time = create_time

class Logistics:
    """物流信息模型"""
    def __init__(self, order_id: str, carrier: str, tracking_no: str,
                 status: str, location: str, update_time: str):
        self.order_id = order_id
        self.carrier = carrier
        self.tracking_no = tracking_no
        self.status = status
        self.location = location
        self.update_time = update_time

# 创建测试数据
order = Order(
    order_id="ORD1234567890",
    user_id="USER001",
    status="已发货",
    amount=299.00,
    items=["iPhone 数据线 x1", "手机壳 x2"],
    create_time="2026-07-05 10:30:00"
)

logistics = Logistics(
    order_id="ORD1234567890",
    carrier="顺丰速运",
    tracking_no="SF1234567890",
    status="运输中",
    location="北京市朝阳区分拣中心",
    update_time="2026-07-09 08:30:00"
)

print(f"✓ Order: {order.order_id}, {order.status}, ¥{order.amount}")
print(f"✓ Logistics: {logistics.carrier}, {logistics.tracking_no}, {logistics.status}")

assert order.order_id == "ORD1234567890"
assert order.amount == 299.00
assert len(order.items) == 2

print("✅ 客服 Agent 数据模型测试通过")


# ============================================
# 测试 5：工具函数业务逻辑
# ============================================
print("\n【测试 5】工具函数业务逻辑")
print("-" * 60)

# 模拟数据库
ORDERS_DB = {
    "ORD1234567890": Order(
        order_id="ORD1234567890",
        user_id="USER001",
        status="已发货",
        amount=299.00,
        items=["iPhone 数据线 x1", "手机壳 x2"],
        create_time="2026-07-05 10:30:00"
    )
}

def query_order(order_id: str) -> str:
    """查询订单详情"""
    if not order_id.startswith("ORD") or len(order_id) != 13:
        return "❌ 订单号格式错误"

    order = ORDERS_DB.get(order_id)
    if not order:
        return f"❌ 订单 {order_id} 不存在"

    return (
        f"📦 订单详情：\n"
        f"  订单号：{order.order_id}\n"
        f"  状态：{order.status}\n"
        f"  金额：¥{order.amount:.2f}"
    )

# 测试存在的订单
result1 = query_order("ORD1234567890")
assert "📦 订单详情" in result1
assert "299.00" in result1
print(f"✓ 查询存在的订单:\n{result1}")

# 测试不存在的订单
result2 = query_order("ORD9999999999")
assert "不存在" in result2
print(f"✓ 查询不存在的订单: {result2}")

# 测试格式错误
result3 = query_order("INVALID")
assert "格式错误" in result3
print(f"✓ 格式错误的订单号: {result3}")

print("✅ 工具函数业务逻辑测试通过")


# ============================================
# 测试 6：Tools 定义格式
# ============================================
print("\n【测试 6】OpenAI Tools 定义格式")
print("-" * 60)

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

print(f"✓ Tools 定义数量: {len(tools)}")
print(f"✓ 第一个工具名称: {tools[0]['function']['name']}")
print(f"✓ 参数类型: {tools[0]['function']['parameters']['type']}")
print(f"✓ 必填参数: {tools[0]['function']['parameters']['required']}")

assert tools[0]['type'] == 'function'
assert tools[0]['function']['name'] == 'get_weather'
assert 'city' in tools[0]['function']['parameters']['required']

print("✅ Tools 定义格式测试通过")


# ============================================
# 测试 7：Pydantic Schema 生成
# ============================================
print("\n【测试 7】Pydantic Schema 生成")
print("-" * 60)

class TransferParams(BaseModel):
    """转账参数"""
    from_account: str = Field(..., description="转出账户")
    to_account: str = Field(..., description="转入账户")
    amount: float = Field(..., gt=0, le=10000, description="转账金额")

# 生成 JSON Schema
schema = TransferParams.model_json_schema()

print(f"✓ Schema type: {schema.get('type')}")
print(f"✓ Properties: {list(schema.get('properties', {}).keys())}")
print(f"✓ Required: {schema.get('required')}")

assert schema['type'] == 'object'
assert 'from_account' in schema['properties']
assert 'to_account' in schema['properties']
assert 'amount' in schema['properties']
assert set(schema['required']) == {'from_account', 'to_account', 'amount'}

print("✅ Pydantic Schema 生成测试通过")


# ============================================
# 总结
# ============================================
print("\n" + "=" * 80)
print("测试总结")
print("=" * 80)

test_results = [
    ("基础工具函数", True),
    ("Pydantic 参数校验", True),
    ("风险等级枚举", True),
    ("客服 Agent 数据模型", True),
    ("工具函数业务逻辑", True),
    ("OpenAI Tools 定义格式", True),
    ("Pydantic Schema 生成", True),
]

for test_name, passed in test_results:
    status = "✅ PASS" if passed else "❌ FAIL"
    print(f"{status} - {test_name}")

all_passed = all(result[1] for result in test_results)

print("\n" + "=" * 80)
if all_passed:
    print("🎉 所有测试通过！教程 02 代码验证完成。")
    print("=" * 80)
    print("\n📝 下一步：")
    print("  1. 配置 API Key (创建 .env 文件)")
    print("  2. 运行示例: python examples/02-tool-calling/01_basic_tool.py")
    print("  3. 阅读教程: tutorial-02-tool-calling.md")
    sys.exit(0)
else:
    print("❌ 部分测试失败，请检查代码")
    sys.exit(1)