"""
Pytest 配置和共享 fixtures
提供测试数据共享机制：
- test_config: 测试配置
- default_app_token: 默认测试应用（从环境变量读取）
- test_table_id: 测试数据表（session级别）
- test_record_id: 单条测试记录（function级别）
- test_record_ids: 批量测试记录（function级别）
"""
import pytest
import json
import time
import os
from yuppie_mcp_feishu.client import get_client
from yuppie_mcp_feishu.impl import (
create_bitable_table,
create_bitable_record,
batch_create_bitable_records,
)
# MCP SDK imports
from mcp.shared.memory import create_connected_server_and_client_session
from yuppie_mcp_feishu.server import server
class TestConfig:
"""测试配置类"""
def __init__(self):
# 默认测试应用 app_token（用于数据表和记录测试）
self.default_app_token = os.getenv("FEISHU_DEFAULT_APP_TOKEN", "")
@pytest.fixture(scope="session")
def test_config() -> TestConfig:
"""
测试配置 fixture
返回 TestConfig 对象，包含 default_app_token
"""
config = TestConfig()
# 如果没有配置默认应用，提示用户
if not config.default_app_token:
pytest.skip(
"FEISHU_DEFAULT_APP_TOKEN not set - "
"please set it in .env to run table and record tests"
)
return config
@pytest.fixture(scope="session")
def default_app_token(test_config: TestConfig) -> str:
"""
默认测试应用 app_token
从环境变量 FEISHU_DEFAULT_APP_TOKEN 读取，用于数据表和记录测试。
避免每次测试都创建新应用。
Scope: session - 整个测试会话使用同一个应用
Returns:
str: 应用 app_token
"""
app_token = test_config.default_app_token
print(f"\n✅ Using default test app: {app_token}")
return app_token
@pytest.fixture(scope="session")
def test_table_id(default_app_token: str) -> str:
"""
创建测试用数据表
Scope: session - 整个测试会话只创建一次
依赖: default_app_token fixture（使用默认应用）
Returns:
str: 数据表 table_id
"""
timestamp = int(time.time())
table_name = f"测试数据表_{timestamp}"
result = create_bitable_table(
app_token=default_app_token,
table_name=table_name,
fields=None, # 使用默认字段
)
data = json.loads(result)
# 尝试获取创建的表 ID
table_id = data.get("table_id")
# 如果创建失败，尝试获取默认表
if not table_id:
from lark_oapi.api.bitable.v1 import ListAppTableRequest
client = get_client()
request = ListAppTableRequest.builder().app_token(default_app_token).build()
response = client.bitable.v1.app_table.list(request)
if response.success():
from lark_oapi import JSON as LarkJSON
response_data = json.loads(LarkJSON.marshal(response.data))
tables = response_data.get("items", [])
if tables:
table_id = tables[0].get("table_id")
print(f"\n✅ Using default table: {table_id}")
else:
pytest.fail("No tables available in test app")
else:
pytest.fail(f"Failed to list tables: {response.msg}")
else:
print(f"\n✅ Created test table: {table_name} ({table_id})")
return table_id
@pytest.fixture(scope="function")
def test_record_id(default_app_token: str, test_table_id: str) -> str:
"""
创建测试用记录
Scope: function - 每个测试函数都会创建新记录
依赖: default_app_token, test_table_id fixtures
Returns:
str: 记录 record_id
"""
result = create_bitable_record(
app_token=default_app_token,
table_id=test_table_id,
fields={}, # 空字段
)
data = json.loads(result)
# 检查是否成功
if not data.get("success", True):
pytest.skip(f"Failed to create test record: {data.get('msg')}")
record_id = data.get("record", {}).get("record_id")
if not record_id:
pytest.skip("No record_id returned")
return record_id
@pytest.fixture(scope="function")
def test_record_ids(default_app_token: str, test_table_id: str, count: int = 3) -> list:
"""
批量创建测试记录
Scope: function - 每个测试函数都会创建新的记录批次
依赖: default_app_token, test_table_id fixtures
Args:
count: 创建的记录数量（默认3条）
Returns:
list: record_id 列表
"""
records = [{"fields": {}} for _ in range(count)]
result = batch_create_bitable_records(
app_token=default_app_token,
table_id=test_table_id,
records=records,
)
data = json.loads(result)
# 检查是否成功
if not data.get("success", True):
pytest.skip(f"Failed to batch create records: {data.get('msg')}")
record_ids = [
item.get("record_id")
for item in data.get("records", [])
if item.get("record_id")
]
if not record_ids:
pytest.skip("No record_ids returned")
return record_ids
# ===== MCP 测试 Fixtures =====
@pytest.fixture
def mcp_server():
"""
MCP Server 实例
返回 yuppie_mcp_feishu 的 MCP server 对象，用于创建测试会话。
Returns:
Server: MCP server 实例
"""
return server
@pytest.fixture
async def client_session(mcp_server):
"""
MCP 客户端会话
使用内存传输创建连接的客户端会话，用于测试 MCP 工具调用。
Args:
mcp_server: MCP server 实例（由 mcp_server fixture 提供）
Yields:
ClientSession: 已初始化的客户端会话
Example:
result = await client_session.call_tool(
name="create_bitable_record",
arguments={"app_token": "xxx", "table_id": "xxx", "fields": {}}
)
"""
async with create_connected_server_and_client_session(mcp_server) as session:
yield session
