PythonGraphQL测试怎么进行

wen python案例 26

本文目录导读:

PythonGraphQL测试怎么进行

  1. 基础测试方法
  2. 使用专业测试库
  3. 高级测试场景
  4. 集成测试配置
  5. 测试报告和监控
  6. 最佳实践

我来详细说明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服务,从简单查询到复杂的认证和性能测试,根据项目需求选择合适的测试策略和工具。

抱歉,评论功能暂时关闭!