本文目录导读:

我来详细说明Python GraphQL测试的方法和最佳实践。
基础测试方法
使用 requests 库直接测试
import requests
import json
# 基本查询测试
def test_graphql_query():
url = "http://localhost:8000/graphql"
query = """
query {
users {
id
name
email
}
}
"""
response = requests.post(
url,
json={"query": query},
headers={"Content-Type": "application/json"}
)
assert response.status_code == 200
data = response.json()
assert "data" in data
assert "users" in data["data"]
# 带变量的查询
def test_query_with_variables():
url = "http://localhost:8000/graphql"
query = """
query GetUser($id: ID!) {
user(id: $id) {
name
email
}
}
"""
variables = {"id": "1"}
response = requests.post(
url,
json={"query": query, "variables": variables}
)
assert response.status_code == 200
data = response.json()
assert data["data"]["user"]["name"] == "张三"
使用 pytest 框架
import pytest
import requests
class TestGraphQL:
BASE_URL = "http://localhost:8000/graphql"
@pytest.fixture
def client(self):
"""创建测试客户端"""
return requests.Session()
def test_get_users(self, client):
query = """
query {
users {
id
name
}
}
"""
response = client.post(
self.BASE_URL,
json={"query": query}
)
assert response.status_code == 200
assert len(response.json()["data"]["users"]) > 0
def test_create_user(self, client):
mutation = """
mutation CreateUser($input: UserInput!) {
createUser(input: $input) {
id
name
email
}
}
"""
variables = {
"input": {
"name": "李四",
"email": "lisi@example.com"
}
}
response = client.post(
self.BASE_URL,
json={"query": mutation, "variables": variables}
)
assert response.status_code == 200
assert response.json()["data"]["createUser"]["name"] == "李四"
使用专业测试库
Graphene-Pytest(推荐用于Django/Flask)
# 安装:pip install graphene-pytest
import pytest
from graphene_pytest import GraphQLClient
@pytest.fixture
def graphql_client():
return GraphQLClient("http://localhost:8000/graphql")
def test_user_query(graphql_client):
query = """
query {
users {
name
email
}
}
"""
result = graphql_client.execute(query)
assert result.status_code == 200
assert len(result.data["users"]) > 0
def test_mutation(graphql_client):
mutation = """
mutation {
createUser(name: "王五", email: "wangwu@example.com") {
id
name
}
}
"""
result = graphql_client.execute(mutation)
assert result.data["createUser"]["name"] == "王五"
使用 sgqlc(GraphQL客户端库)
# 安装:pip install sgqlc
from sgqlc.endpoint.requests import RequestsEndpoint
from sgqlc.operation import Operation
from schema import Schema # 你的schema定义
def test_with_sgqlc():
endpoint = RequestsEndpoint("http://localhost:8000/graphql")
op = Operation(Schema.QueryType)
users = op.users()
users.id()
users.name()
users.email()
data = endpoint(op)
assert data is not None
assert "users" in data
高级测试场景
认证和权限测试
import pytest
import requests
class TestAuthenticatedGraphQL:
BASE_URL = "http://localhost:8000/graphql"
def get_auth_token(self):
# 获取认证token
login_mutation = """
mutation {
login(email: "admin@example.com", password: "password123") {
token
}
}
"""
response = requests.post(
self.BASE_URL,
json={"query": login_mutation}
)
return response.json()["data"]["login"]["token"]
def test_authenticated_query(self):
token = self.get_auth_token()
query = """
query {
protectedData {
id
secret
}
}
"""
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
response = requests.post(
self.BASE_URL,
json={"query": query},
headers=headers
)
assert response.status_code == 200
def test_unauthorized_access(self):
query = """
query {
protectedData {
id
}
}
"""
response = requests.post(
self.BASE_URL,
json={"query": query}
)
# 应该返回错误
assert response.status_code == 401 or "errors" in response.json()
批量测试和参数化
import pytest
import requests
class TestGraphQLBatch:
BASE_URL = "http://localhost:8000/graphql"
@pytest.mark.parametrize("user_id, expected_name", [
("1", "张三"),
("2", "李四"),
("3", "王五"),
])
def test_multiple_users(self, user_id, expected_name):
query = f"""
query {{
user(id: "{user_id}") {{
name
}}
}}
"""
response = requests.post(
self.BASE_URL,
json={"query": query}
)
assert response.json()["data"]["user"]["name"] == expected_name
@pytest.mark.parametrize("query, expected_error", [
("{ invalidField }", "Cannot query field"),
("query { user(id: \"invalid\") { name } }", "not found"),
])
def test_error_cases(self, query, expected_error):
response = requests.post(
self.BASE_URL,
json={"query": query}
)
errors = response.json().get("errors", [])
assert any(expected_error in str(error) for error in errors)
集成测试配置
Docker 测试环境
# test_docker_setup.py
import subprocess
import time
import requests
import pytest
@pytest.fixture(scope="module")
def docker_compose():
"""启动Docker测试环境"""
subprocess.run(["docker-compose", "up", "-d"])
time.sleep(10) # 等待服务启动
yield
subprocess.run(["docker-compose", "down"])
def test_graphql_in_docker(docker_compose):
response = requests.get("http://localhost:8000/graphql")
assert response.status_code == 200
数据库测试
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
@pytest.fixture
def db_session():
# 使用内存数据库
engine = create_engine("sqlite:///:memory:")
SessionLocal = sessionmaker(bind=engine)
session = SessionLocal()
# 创建表和数据
Base.metadata.create_all(bind=engine)
yield session
session.close()
def test_graphql_with_db(db_session):
# 测试数据库交互
pass
测试报告和监控
import pytest
import json
from datetime import datetime
class GraphQLTestReport:
def __init__(self):
self.results = []
def record_test(self, test_name, query, status, response_time):
self.results.append({
"test_name": test_name,
"query": query,
"status": status,
"response_time": response_time,
"timestamp": datetime.now().isoformat()
})
def generate_report(self):
report = {
"total_tests": len(self.results),
"passed": sum(1 for r in self.results if r["status"] == "passed"),
"failed": sum(1 for r in self.results if r["status"] == "failed"),
"average_response_time": sum(r["response_time"] for r in self.results) / len(self.results),
"tests": self.results
}
with open("graphql_test_report.json", "w") as f:
json.dump(report, f, indent=2)
# 在测试中使用
report = GraphQLTestReport()
def test_with_reporting():
start_time = time.time()
# 执行测试
response_time = time.time() - start_time
report.record_test(
"test_get_users",
"{ users { name } }",
"passed",
response_time
)
最佳实践
# test_best_practices.py
import pytest
import requests
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class GraphQLTester:
def __init__(self, base_url):
self.base_url = base_url
self.session = requests.Session()
def execute_query(self, query, variables=None, headers=None):
payload = {"query": query}
if variables:
payload["variables"] = variables
response = self.session.post(
self.base_url,
json=payload,
headers=headers or {}
)
logger.info(f"Query executed: {query[:50]}...")
return response
def assert_no_errors(self, response):
data = response.json()
assert "errors" not in data, f"GraphQL errors: {data.get('errors')}"
def assert_has_data(self, response, field_path):
data = response.json()["data"]
fields = field_path.split(".")
for field in fields:
data = data[field]
assert data is not None
# 使用示例
def test_comprehensive():
tester = GraphQLTester("http://localhost:8000/graphql")
# 测试查询
query = "{ users { id name } }"
response = tester.execute_query(query)
tester.assert_no_errors(response)
tester.assert_has_data(response, "users")
# 测试错误处理
query = "{ invalidField }"
response = tester.execute_query(query)
assert "errors" in response.json()
# 测试性能
import time
start = time.time()
for _ in range(10):
tester.execute_query("{ users { id } }")
duration = time.time() - start
assert duration < 5.0 # 10次查询应在5秒内
这些方法可以帮助你全面测试Python GraphQL服务,从简单查询到复杂的认证和性能测试,根据项目需求选择合适的测试策略和工具。