16.pytest的requests

作者

Pytest 与 Requests 详解

1. 什么是 Requests

1.1 Requests 简介

Requests 是 Python 中最受欢迎的 HTTP 库之一，它让发送 HTTP 请求变得非常简单和直观。Requests 库的设计哲学是”为人类而设计”，提供了简洁优雅的 API。

1.2 Requests 的特点

  • 简单易用：API 设计直观，学习成本低
  • 功能强大：支持所有 HTTP 方法（GET、POST、PUT、DELETE 等）
  • 自动处理：自动处理 URL 编码、连接池、会话等
  • 广泛使用：在 Python 社区中被广泛采用
  • 文档完善：拥有详细的中文和英文文档

1.3 Requests 在测试中的作用

在 pytest 测试中，Requests 主要用于：

  1. API 接口测试：测试 RESTful API 的各种接口
  2. HTTP 请求模拟：模拟客户端发送 HTTP 请求
  3. 数据验证：验证 API 返回的数据格式和内容
  4. 集成测试：测试多个服务之间的集成
  5. 性能测试：测试 API 的响应时间和性能

1.4 安装 Requests

在开始使用之前，需要先安装 Requests 库：

# 使用 pip 安装
pip install requests

# 使用 pip 安装指定版本
pip install requests==2.31.0

# 使用 conda 安装
conda install requests

验证安装是否成功：

import requests
print(requests.__version__)  # 输出：2.31.0

2. Requests 基础使用

2.1 发送第一个请求

让我们从最简单的 GET 请求开始：

import requests

# 发送 GET 请求
response = requests.get('https://www.baidu.com')

# 查看响应状态码
print(response.status_code)  # 输出：200

# 查看响应内容
print(response.text)  # 输出：HTML 内容

2.2 理解 Response 对象

当我们发送请求后，会得到一个 Response 对象，它包含了服务器返回的所有信息：

import requests

response = requests.get('https://api.github.com')

# 状态码
print(response.status_code)  # 200

# 响应头
print(response.headers)  # 字典格式的响应头

# 响应内容（文本格式）
print(response.text)  # 字符串格式的响应内容

# 响应内容（二进制格式）
print(response.content)  # 字节格式的响应内容

# 响应内容（JSON 格式，如果响应是 JSON）
print(response.json())  # 字典格式的 JSON 数据

# 响应 URL（可能和请求 URL 不同，因为有重定向）
print(response.url)  # 最终响应的 URL

# 响应历史（重定向历史）
print(response.history)  # 重定向的列表

# 响应时间（需要手动计算）
import time
start = time.time()
response = requests.get('https://api.github.com')
elapsed = time.time() - start
print(f"响应时间：{elapsed} 秒")

2.3 在 pytest 中使用 Requests

在 pytest 测试中使用 Requests 的基本模式：

import pytest
import requests

def test_get_request():
    """测试 GET 请求"""
    url = 'https://api.github.com'
    response = requests.get(url)

    # 断言状态码
    assert response.status_code == 200

    # 断言响应内容不为空
    assert response.text is not None

    # 断言响应是 JSON 格式
    data = response.json()
    assert isinstance(data, dict)

3. HTTP 请求方法详解

3.1 GET 请求

GET 请求用于获取资源，是最常用的请求方法：

3.1.1 基本 GET 请求

import pytest
import requests

def test_get_basic():
    """基本 GET 请求"""
    url = 'https://api.github.com/users/octocat'
    response = requests.get(url)

    assert response.status_code == 200
    data = response.json()
    assert 'login' in data
    assert data['login'] == 'octocat'

3.1.2 带查询参数的 GET 请求

查询参数可以通过 params 参数传递：

import pytest
import requests

def test_get_with_params():
    """带查询参数的 GET 请求"""
    url = 'https://api.github.com/search/repositories'

    # 方式一：使用 params 参数（推荐）
    params = {
        'q': 'python',
        'sort': 'stars',
        'order': 'desc'
    }
    response = requests.get(url, params=params)

    assert response.status_code == 200
    data = response.json()
    assert 'items' in data

    # 方式二：直接在 URL 中拼接（不推荐）
    url_with_params = 'https://api.github.com/search/repositories?q=python&sort=stars'
    response2 = requests.get(url_with_params)
    assert response2.status_code == 200

3.1.3 带请求头的 GET 请求

import pytest
import requests

def test_get_with_headers():
    """带请求头的 GET 请求"""
    url = 'https://api.github.com/user'

    headers = {
        'User-Agent': 'MyApp/1.0',
        'Accept': 'application/vnd.github.v3+json'
    }

    response = requests.get(url, headers=headers)
    # 注意：这个接口需要认证，所以会返回 401
    assert response.status_code == 401

3.2 POST 请求

POST 请求用于创建资源或提交数据：

3.2.1 发送 JSON 数据

import pytest
import requests

def test_post_json():
    """发送 JSON 数据的 POST 请求"""
    url = 'https://httpbin.org/post'

    data = {
        'name': '张三',
        'age': 25,
        'email': 'zhangsan@example.com'
    }

    # 使用 json 参数，会自动设置 Content-Type 为 application/json
    response = requests.post(url, json=data)

    assert response.status_code == 200
    result = response.json()
    assert result['json']['name'] == '张三'
    assert result['json']['age'] == 25

3.2.2 发送表单数据

import pytest
import requests

def test_post_form():
    """发送表单数据的 POST 请求"""
    url = 'https://httpbin.org/post'

    form_data = {
        'username': 'testuser',
        'password': 'testpass'
    }

    # 使用 data 参数发送表单数据
    response = requests.post(url, data=form_data)

    assert response.status_code == 200
    result = response.json()
    assert result['form']['username'] == 'testuser'

3.2.3 发送文件

import pytest
import requests
import os

def test_post_file():
    """发送文件的 POST 请求"""
    url = 'https://httpbin.org/post'

    # 创建一个测试文件
    test_file_path = 'test_file.txt'
    with open(test_file_path, 'w', encoding='utf-8') as f:
        f.write('这是一个测试文件')

    try:
        # 发送文件
        with open(test_file_path, 'rb') as f:
            files = {'file': f}
            response = requests.post(url, files=files)

        assert response.status_code == 200
        result = response.json()
        assert 'files' in result
    finally:
        # 清理测试文件
        if os.path.exists(test_file_path):
            os.remove(test_file_path)

3.2.4 发送原始数据

import pytest
import requests

def test_post_raw_data():
    """发送原始数据的 POST 请求"""
    url = 'https://httpbin.org/post'

    raw_data = '这是一段原始文本数据'

    response = requests.post(
        url,
        data=raw_data,
        headers={'Content-Type': 'text/plain'}
    )

    assert response.status_code == 200
    result = response.json()
    assert result['data'] == raw_data

3.3 PUT 请求

PUT 请求用于更新资源：

import pytest
import requests

def test_put_request():
    """PUT 请求示例"""
    url = 'https://httpbin.org/put'

    data = {
        'id': 1,
        'name': '更新后的名称',
        'status': 'active'
    }

    response = requests.put(url, json=data)

    assert response.status_code == 200
    result = response.json()
    assert result['json']['name'] == '更新后的名称'

3.4 DELETE 请求

DELETE 请求用于删除资源：

import pytest
import requests

def test_delete_request():
    """DELETE 请求示例"""
    url = 'https://httpbin.org/delete'

    response = requests.delete(url)

    assert response.status_code == 200
    result = response.json()
    assert 'url' in result

3.5 PATCH 请求

PATCH 请求用于部分更新资源：

import pytest
import requests

def test_patch_request():
    """PATCH 请求示例"""
    url = 'https://httpbin.org/patch'

    data = {
        'status': 'inactive'
    }

    response = requests.patch(url, json=data)

    assert response.status_code == 200
    result = response.json()
    assert result['json']['status'] == 'inactive'

3.6 HEAD 请求

HEAD 请求只获取响应头，不获取响应体：

import pytest
import requests

def test_head_request():
    """HEAD 请求示例"""
    url = 'https://httpbin.org/get'

    response = requests.head(url)

    assert response.status_code == 200
    # HEAD 请求不返回响应体
    assert response.text == ''
    # 但可以获取响应头
    assert 'Content-Type' in response.headers

3.7 OPTIONS 请求

OPTIONS 请求用于获取服务器支持的 HTTP 方法：

import pytest
import requests

def test_options_request():
    """OPTIONS 请求示例"""
    url = 'https://httpbin.org/get'

    response = requests.options(url)

    assert response.status_code == 200
    # 查看允许的方法
    allowed_methods = response.headers.get('Allow', '')
    print(f"允许的方法：{allowed_methods}")

4. 请求参数详解

4.1 URL 参数

URL 参数可以通过 params 参数传递，Requests 会自动进行 URL 编码：

import pytest
import requests

def test_url_params():
    """URL 参数示例"""
    url = 'https://httpbin.org/get'

    params = {
        'key1': 'value1',
        'key2': 'value2',
        'key3': ['value3a', 'value3b']  # 多个值
    }

    response = requests.get(url, params=params)

    assert response.status_code == 200
    result = response.json()
    assert result['args']['key1'] == 'value1'
    assert result['args']['key2'] == 'value2'
    # 多个值会变成列表
    assert 'value3a' in result['args']['key3']

4.2 请求头（Headers）

请求头可以通过 headers 参数传递：

import pytest
import requests

def test_custom_headers():
    """自定义请求头示例"""
    url = 'https://httpbin.org/headers'

    headers = {
        'User-Agent': 'MyTestApp/1.0',
        'Accept': 'application/json',
        'X-Custom-Header': 'custom-value',
        'Authorization': 'Bearer token123'
    }

    response = requests.get(url, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result['headers']['User-Agent'] == 'MyTestApp/1.0'
    assert result['headers']['X-Custom-Header'] == 'custom-value'

4.3 请求体（Body）

请求体可以通过 data、json、files 等参数传递：

import pytest
import requests

def test_request_body():
    """请求体示例"""
    url = 'https://httpbin.org/post'

    # JSON 数据
    json_data = {'name': 'test', 'value': 123}

    # 表单数据
    form_data = {'field1': 'value1', 'field2': 'value2'}

    # 发送 JSON
    response_json = requests.post(url, json=json_data)
    assert response_json.status_code == 200

    # 发送表单
    response_form = requests.post(url, data=form_data)
    assert response_form.status_code == 200

4.4 超时设置

设置请求超时时间，避免请求无限等待：

import pytest
import requests

def test_timeout():
    """超时设置示例"""
    url = 'https://httpbin.org/delay/2'

    # 设置超时时间为 1 秒
    try:
        response = requests.get(url, timeout=1)
    except requests.exceptions.Timeout:
        print("请求超时")
        assert True  # 预期会超时

    # 设置连接超时和读取超时
    try:
        response = requests.get(
            url,
            timeout=(3, 5)  # 连接超时 3 秒，读取超时 5 秒
        )
        assert response.status_code == 200
    except requests.exceptions.Timeout:
        print("请求超时")

4.5 重定向处理

默认情况下，Requests 会自动处理重定向：

import pytest
import requests

def test_redirect():
    """重定向处理示例"""
    # 这个 URL 会重定向到 https://httpbin.org/get
    url = 'https://httpbin.org/redirect/1'

    # 默认允许重定向
    response = requests.get(url)
    assert response.status_code == 200
    assert len(response.history) == 1  # 有 1 次重定向

    # 禁止重定向
    response_no_redirect = requests.get(url, allow_redirects=False)
    assert response_no_redirect.status_code == 302  # 重定向状态码
    assert len(response_no_redirect.history) == 0

5. 响应处理详解

5.1 状态码检查

检查响应状态码是最基本的验证：

import pytest
import requests

def test_status_code():
    """状态码检查示例"""
    url = 'https://httpbin.org/status/200'

    response = requests.get(url)

    # 方式一：直接检查状态码
    assert response.status_code == 200

    # 方式二：使用 raise_for_status() 抛出异常
    response.raise_for_status()  # 如果状态码不是 2xx，会抛出异常

    # 方式三：检查状态码范围
    assert 200 <= response.status_code < 300

5.2 响应内容解析

根据响应内容类型，使用不同的方法解析：

import pytest
import requests

def test_response_content():
    """响应内容解析示例"""
    url = 'https://httpbin.org/json'

    response = requests.get(url)

    # JSON 响应
    if 'application/json' in response.headers.get('Content-Type', ''):
        data = response.json()
        assert isinstance(data, dict)

    # 文本响应
    text_content = response.text
    assert isinstance(text_content, str)

    # 二进制响应
    binary_content = response.content
    assert isinstance(binary_content, bytes)

    # 原始响应（用于流式传输）
    raw_response = response.raw
    assert raw_response is not None

5.3 响应头检查

检查响应头中的信息：

import pytest
import requests

def test_response_headers():
    """响应头检查示例"""
    url = 'https://httpbin.org/get'

    response = requests.get(url)

    # 获取单个响应头
    content_type = response.headers.get('Content-Type')
    assert 'application/json' in content_type

    # 获取所有响应头
    all_headers = response.headers
    assert 'Content-Type' in all_headers

    # 检查响应头（不区分大小写）
    assert 'content-type' in response.headers
    assert 'Content-Type' in response.headers

5.4 响应时间测量

测量请求的响应时间：

import pytest
import requests
import time

def test_response_time():
    """响应时间测量示例"""
    url = 'https://httpbin.org/delay/1'

    start_time = time.time()
    response = requests.get(url)
    elapsed_time = time.time() - start_time

    assert response.status_code == 200
    assert elapsed_time >= 1.0  # 至少需要 1 秒
    print(f"响应时间：{elapsed_time:.2f} 秒")

6. 认证和授权

6.1 Basic 认证

Basic 认证是最简单的 HTTP 认证方式：

import pytest
import requests
from requests.auth import HTTPBasicAuth

def test_basic_auth():
    """Basic 认证示例"""
    url = 'https://httpbin.org/basic-auth/user/pass'

    # 方式一：使用 HTTPBasicAuth
    response = requests.get(
        url,
        auth=HTTPBasicAuth('user', 'pass')
    )
    assert response.status_code == 200

    # 方式二：使用元组（更简洁）
    response2 = requests.get(url, auth=('user', 'pass'))
    assert response2.status_code == 200

6.2 Digest 认证

Digest 认证比 Basic 认证更安全：

import pytest
import requests
from requests.auth import HTTPDigestAuth

def test_digest_auth():
    """Digest 认证示例"""
    url = 'https://httpbin.org/digest-auth/auth/user/pass'

    response = requests.get(
        url,
        auth=HTTPDigestAuth('user', 'pass')
    )
    assert response.status_code == 200

6.3 Bearer Token 认证

Bearer Token 是 API 中最常用的认证方式：

import pytest
import requests

def test_bearer_token():
    """Bearer Token 认证示例"""
    url = 'https://api.example.com/protected'

    token = 'your_token_here'

    # 方式一：在请求头中设置
    headers = {
        'Authorization': f'Bearer {token}'
    }
    response = requests.get(url, headers=headers)

    # 方式二：使用自定义认证类
    class BearerAuth(requests.auth.AuthBase):
        def __init__(self, token):
            self.token = token

        def __call__(self, r):
            r.headers['Authorization'] = f'Bearer {self.token}'
            return r

    response2 = requests.get(url, auth=BearerAuth(token))

6.4 API Key 认证

API Key 通常放在请求头或查询参数中：

import pytest
import requests

def test_api_key():
    """API Key 认证示例"""
    url = 'https://api.example.com/data'

    api_key = 'your_api_key_here'

    # 方式一：放在请求头中
    headers = {
        'X-API-Key': api_key
    }
    response = requests.get(url, headers=headers)

    # 方式二：放在查询参数中
    params = {
        'api_key': api_key
    }
    response2 = requests.get(url, params=params)

6.5 在 pytest 中使用认证 Fixture

将认证逻辑封装成 fixture，方便复用：

import pytest
import requests

@pytest.fixture(scope="session")
def auth_token():
    """获取认证 token"""
    login_url = 'https://api.example.com/login'
    login_data = {
        'username': 'testuser',
        'password': 'testpass'
    }

    response = requests.post(login_url, json=login_data)
    response.raise_for_status()

    token = response.json()['token']
    return token

@pytest.fixture(scope="session")
def auth_headers(auth_token):
    """认证请求头"""
    return {
        'Authorization': f'Bearer {auth_token}',
        'Content-Type': 'application/json'
    }

def test_protected_endpoint(auth_headers):
    """测试需要认证的接口"""
    url = 'https://api.example.com/protected'
    response = requests.get(url, headers=auth_headers)

    assert response.status_code == 200

7. 会话管理（Session）

7.1 什么是 Session

Session 可以保持 cookies 和连接，在多个请求之间共享：

import pytest
import requests

def test_session_basic():
    """Session 基本使用"""
    # 创建 Session 对象
    session = requests.Session()

    # 设置默认请求头
    session.headers.update({'User-Agent': 'MyApp/1.0'})

    # 发送请求
    response1 = session.get('https://httpbin.org/cookies/set/sessionid/12345')
    response2 = session.get('https://httpbin.org/cookies')

    # Session 会自动保持 cookies
    cookies = response2.json()['cookies']
    assert cookies['sessionid'] == '12345'

    # 关闭 Session
    session.close()

7.2 Session 的优势

使用 Session 的优势：

  1. 保持连接：复用 TCP 连接，提高性能
  2. 保持 Cookies：自动管理 cookies
  3. 共享配置：可以设置默认的 headers、auth 等
import pytest
import requests

def test_session_advantages():
    """Session 优势示例"""
    session = requests.Session()

    # 设置默认配置
    session.headers.update({
        'User-Agent': 'MyApp/1.0',
        'Accept': 'application/json'
    })

    # 设置默认认证
    session.auth = ('user', 'pass')

    # 设置默认超时
    session.timeout = 5

    # 所有请求都会使用这些默认配置
    response = session.get('https://httpbin.org/headers')
    assert response.status_code == 200

    session.close()

7.3 在 pytest 中使用 Session Fixture

将 Session 封装成 fixture：

import pytest
import requests

@pytest.fixture(scope="session")
def api_session():
    """创建 API Session"""
    session = requests.Session()

    # 设置基础 URL
    session.base_url = 'https://api.example.com'

    # 设置默认请求头
    session.headers.update({
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    })

    yield session

    # 清理
    session.close()

def test_with_session(api_session):
    """使用 Session 的测试"""
    # 使用相对路径
    response = api_session.get('/users')
    assert response.status_code == 200

    # Session 会自动添加 base_url
    # 实际请求的是 https://api.example.com/users

8. Cookies 管理

8.1 获取 Cookies

import pytest
import requests

def test_get_cookies():
    """获取 Cookies 示例"""
    url = 'https://httpbin.org/cookies/set/testcookie/testvalue'

    response = requests.get(url)

    # 获取所有 cookies
    cookies = response.cookies
    print(f"Cookies: {cookies}")

    # 获取特定 cookie
    cookie_value = response.cookies.get('testcookie')
    print(f"Cookie 值: {cookie_value}")

8.2 发送 Cookies

import pytest
import requests

def test_send_cookies():
    """发送 Cookies 示例"""
    url = 'https://httpbin.org/cookies'

    # 方式一：使用 cookies 参数
    cookies = {
        'cookie1': 'value1',
        'cookie2': 'value2'
    }
    response = requests.get(url, cookies=cookies)

    assert response.status_code == 200
    result = response.json()
    assert result['cookies']['cookie1'] == 'value1'

    # 方式二：使用 Session（推荐）
    session = requests.Session()
    session.cookies.update(cookies)
    response2 = session.get(url)
    assert response2.status_code == 200

8.3 CookieJar 对象

Cookies 存储在 CookieJar 对象中：

import pytest
import requests
from http.cookiejar import CookieJar

def test_cookiejar():
    """CookieJar 示例"""
    jar = CookieJar()

    session = requests.Session()
    session.cookies = jar

    # 设置 cookie
    response = session.get('https://httpbin.org/cookies/set/sessionid/12345')

    # 查看 cookies
    for cookie in jar:
        print(f"Cookie: {cookie.name} = {cookie.value}")

9. 错误处理和异常

9.1 Requests 异常类型

Requests 库定义了多种异常类型：

import pytest
import requests

def test_exceptions():
    """异常类型示例"""
    # ConnectionError：网络连接错误
    try:
        requests.get('http://nonexistent-domain-12345.com', timeout=1)
    except requests.exceptions.ConnectionError as e:
        print(f"连接错误：{e}")

    # Timeout：请求超时
    try:
        requests.get('https://httpbin.org/delay/10', timeout=1)
    except requests.exceptions.Timeout as e:
        print(f"超时错误：{e}")

    # HTTPError：HTTP 错误（4xx, 5xx）
    try:
        response = requests.get('https://httpbin.org/status/404')
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print(f"HTTP 错误：{e}")

    # RequestException：所有请求异常的基类
    try:
        requests.get('invalid-url')
    except requests.exceptions.RequestException as e:
        print(f"请求异常：{e}")

9.2 错误处理最佳实践

在测试中正确处理错误：

import pytest
import requests

def test_error_handling():
    """错误处理最佳实践"""
    url = 'https://api.example.com/data'

    try:
        response = requests.get(url, timeout=5)
        response.raise_for_status()  # 如果不是 2xx，抛出异常

        data = response.json()
        assert data is not None

    except requests.exceptions.Timeout:
        pytest.fail("请求超时")
    except requests.exceptions.ConnectionError:
        pytest.fail("连接失败")
    except requests.exceptions.HTTPError as e:
        pytest.fail(f"HTTP 错误：{e}")
    except requests.exceptions.RequestException as e:
        pytest.fail(f"请求异常：{e}")
    except ValueError:
        pytest.fail("响应不是有效的 JSON")

9.3 重试机制

实现请求重试机制：

import pytest
import requests
import time

def test_retry_mechanism():
    """重试机制示例"""
    url = 'https://api.example.com/data'
    max_retries = 3
    retry_delay = 1

    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()

            # 成功，退出循环
            assert response.status_code == 200
            break

        except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
            if attempt < max_retries - 1:
                print(f"尝试 {attempt + 1} 失败，{retry_delay} 秒后重试...")
                time.sleep(retry_delay)
            else:
                pytest.fail(f"重试 {max_retries} 次后仍然失败：{e}")

9.4 使用 requests.adapters.HTTPAdapter 实现自动重试

更优雅的重试方式：

import pytest
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def test_auto_retry():
    """自动重试示例"""
    session = requests.Session()

    # 配置重试策略
    retry_strategy = Retry(
        total=3,  # 总共重试 3 次
        backoff_factor=1,  # 重试间隔
        status_forcelist=[429, 500, 502, 503, 504],  # 需要重试的状态码
    )

    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)

    # 发送请求，会自动重试
    response = session.get('https://httpbin.org/status/500')
    # 注意：这个请求会重试，但最终可能仍然失败

10. 文件上传和下载

10.1 上传文件

import pytest
import requests
import os

def test_upload_file():
    """上传文件示例"""
    url = 'https://httpbin.org/post'

    # 创建测试文件
    test_file_path = 'test_upload.txt'
    with open(test_file_path, 'w', encoding='utf-8') as f:
        f.write('这是要上传的文件内容')

    try:
        # 上传单个文件
        with open(test_file_path, 'rb') as f:
            files = {'file': ('test_upload.txt', f, 'text/plain')}
            response = requests.post(url, files=files)

        assert response.status_code == 200
        result = response.json()
        assert 'files' in result

        # 上传多个文件
        with open(test_file_path, 'rb') as f1, open(test_file_path, 'rb') as f2:
            files = {
                'file1': ('file1.txt', f1, 'text/plain'),
                'file2': ('file2.txt', f2, 'text/plain')
            }
            response2 = requests.post(url, files=files)
            assert response2.status_code == 200

    finally:
        # 清理测试文件
        if os.path.exists(test_file_path):
            os.remove(test_file_path)

10.2 下载文件

import pytest
import requests
import os

def test_download_file():
    """下载文件示例"""
    url = 'https://httpbin.org/bytes/1024'  # 下载 1KB 的数据

    response = requests.get(url, stream=True)
    response.raise_for_status()

    # 保存文件
    download_path = 'downloaded_file.bin'
    try:
        with open(download_path, 'wb') as f:
            # 使用 stream=True 时，需要手动写入
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:
                    f.write(chunk)

        # 验证文件大小
        file_size = os.path.getsize(download_path)
        assert file_size == 1024

    finally:
        # 清理下载的文件
        if os.path.exists(download_path):
            os.remove(download_path)

10.3 流式下载大文件

对于大文件，使用流式下载：

import pytest
import requests
import os

def test_stream_download():
    """流式下载大文件示例"""
    url = 'https://httpbin.org/stream/10'  # 流式响应

    response = requests.get(url, stream=True)
    response.raise_for_status()

    download_path = 'streamed_data.txt'
    try:
        with open(download_path, 'wb') as f:
            # 逐块下载
            for chunk in response.iter_content(chunk_size=1024):
                if chunk:
                    f.write(chunk)
                    # 可以在这里添加进度显示

        assert os.path.exists(download_path)

    finally:
        if os.path.exists(download_path):
            os.remove(download_path)

11. 代理设置

11.1 HTTP 代理

import pytest
import requests

def test_http_proxy():
    """HTTP 代理示例"""
    proxies = {
        'http': 'http://proxy.example.com:8080',
        'https': 'http://proxy.example.com:8080'
    }

    # 使用代理发送请求
    response = requests.get('https://httpbin.org/ip', proxies=proxies)
    assert response.status_code == 200

11.2 需要认证的代理

import pytest
import requests

def test_authenticated_proxy():
    """需要认证的代理示例"""
    proxies = {
        'http': 'http://username:password@proxy.example.com:8080',
        'https': 'http://username:password@proxy.example.com:8080'
    }

    response = requests.get('https://httpbin.org/ip', proxies=proxies)
    assert response.status_code == 200

11.3 在 Session 中设置代理

import pytest
import requests

def test_session_proxy():
    """Session 中设置代理"""
    session = requests.Session()

    proxies = {
        'http': 'http://proxy.example.com:8080',
        'https': 'http://proxy.example.com:8080'
    }

    session.proxies.update(proxies)

    # 所有请求都会使用代理
    response = session.get('https://httpbin.org/ip')
    assert response.status_code == 200

    session.close()

12. SSL 证书验证

12.1 验证 SSL 证书（默认）

默认情况下，Requests 会验证 SSL 证书：

import pytest
import requests

def test_ssl_verification():
    """SSL 证书验证示例"""
    url = 'https://httpbin.org/get'

    # 默认验证证书
    response = requests.get(url, verify=True)
    assert response.status_code == 200

    # 明确指定验证
    response2 = requests.get(url, verify=True)
    assert response2.status_code == 200

12.2 禁用 SSL 证书验证（不推荐）

import pytest
import requests
import urllib3

# 禁用 SSL 警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def test_disable_ssl_verification():
    """禁用 SSL 证书验证（仅用于测试）"""
    url = 'https://self-signed.badssl.com/'

    # 禁用验证（不推荐，仅用于测试环境）
    response = requests.get(url, verify=False)
    # 注意：这会产生安全警告

12.3 使用自定义证书

import pytest
import requests

def test_custom_certificate():
    """使用自定义证书"""
    url = 'https://api.example.com'

    # 指定证书文件
    response = requests.get(
        url,
        verify='/path/to/certificate.pem'
    )
    assert response.status_code == 200

13. 在 pytest 中封装 Requests

13.1 创建 API 客户端类

将 API 请求封装成类，方便管理和复用：

import requests
from typing import Dict, Optional, Any

class APIClient:
    """API 客户端封装类"""

    def __init__(self, base_url: str, timeout: int = 10):
        """
        初始化 API 客户端

        参数:
            base_url: API 基础 URL
            timeout: 默认超时时间（秒）
        """
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        })

    def set_auth_token(self, token: str):
        """设置认证 token"""
        self.session.headers['Authorization'] = f'Bearer {token}'

    def get(self, endpoint: str, params: Optional[Dict] = None, **kwargs) -> requests.Response:
        """发送 GET 请求"""
        url = f"{self.base_url}{endpoint}"
        kwargs.setdefault('timeout', self.timeout)
        return self.session.get(url, params=params, **kwargs)

    def post(self, endpoint: str, data: Optional[Dict] = None, json: Optional[Dict] = None, **kwargs) -> requests.Response:
        """发送 POST 请求"""
        url = f"{self.base_url}{endpoint}"
        kwargs.setdefault('timeout', self.timeout)
        return self.session.post(url, data=data, json=json, **kwargs)

    def put(self, endpoint: str, data: Optional[Dict] = None, json: Optional[Dict] = None, **kwargs) -> requests.Response:
        """发送 PUT 请求"""
        url = f"{self.base_url}{endpoint}"
        kwargs.setdefault('timeout', self.timeout)
        return self.session.put(url, data=data, json=json, **kwargs)

    def delete(self, endpoint: str, **kwargs) -> requests.Response:
        """发送 DELETE 请求"""
        url = f"{self.base_url}{endpoint}"
        kwargs.setdefault('timeout', self.timeout)
        return self.session.delete(url, **kwargs)

    def close(self):
        """关闭 Session"""
        self.session.close()

13.2 在 pytest 中使用 API 客户端

import pytest
from api_client import APIClient

@pytest.fixture(scope="session")
def api_client():
    """创建 API 客户端"""
    client = APIClient(base_url='https://api.example.com')
    yield client
    client.close()

@pytest.fixture(scope="session")
def authenticated_api_client(api_client):
    """已认证的 API 客户端"""
    # 登录获取 token
    login_response = api_client.post('/login', json={
        'username': 'testuser',
        'password': 'testpass'
    })
    token = login_response.json()['token']

    # 设置 token
    api_client.set_auth_token(token)
    return api_client

def test_get_users(authenticated_api_client):
    """获取用户列表"""
    response = authenticated_api_client.get('/users')
    assert response.status_code == 200
    users = response.json()
    assert isinstance(users, list)

def test_create_user(authenticated_api_client):
    """创建用户"""
    user_data = {
        'name': '新用户',
        'email': 'newuser@example.com'
    }
    response = authenticated_api_client.post('/users', json=user_data)
    assert response.status_code == 201
    created_user = response.json()
    assert created_user['name'] == '新用户'

14. 实际测试案例

14.1 案例一：用户管理 API 测试

完整的用户管理 API 测试示例：

import pytest
import requests

BASE_URL = 'https://api.example.com'

@pytest.fixture(scope="session")
def auth_token():
    """获取认证 token"""
    login_url = f'{BASE_URL}/auth/login'
    login_data = {
        'username': 'admin',
        'password': 'admin123'
    }

    response = requests.post(login_url, json=login_data)
    response.raise_for_status()

    token = response.json()['data']['token']
    return token

@pytest.fixture(scope="session")
def auth_headers(auth_token):
    """认证请求头"""
    return {
        'Authorization': f'Bearer {auth_token}',
        'Content-Type': 'application/json'
    }

class TestUserAPI:
    """用户 API 测试类"""

    def test_get_user_list(self, auth_headers):
        """测试获取用户列表"""
        url = f'{BASE_URL}/users'
        response = requests.get(url, headers=auth_headers)

        assert response.status_code == 200
        data = response.json()
        assert 'users' in data
        assert isinstance(data['users'], list)

    def test_get_user_by_id(self, auth_headers):
        """测试根据 ID 获取用户"""
        user_id = 1
        url = f'{BASE_URL}/users/{user_id}'
        response = requests.get(url, headers=auth_headers)

        assert response.status_code == 200
        user = response.json()
        assert user['id'] == user_id
        assert 'name' in user
        assert 'email' in user

    def test_create_user(self, auth_headers):
        """测试创建用户"""
        url = f'{BASE_URL}/users'
        user_data = {
            'name': '测试用户',
            'email': 'test@example.com',
            'password': 'test123'
        }

        response = requests.post(url, json=user_data, headers=auth_headers)

        assert response.status_code == 201
        created_user = response.json()
        assert created_user['name'] == user_data['name']
        assert created_user['email'] == user_data['email']

        # 返回创建的用户 ID，用于后续测试
        return created_user['id']

    def test_update_user(self, auth_headers):
        """测试更新用户"""
        # 先创建一个用户
        create_url = f'{BASE_URL}/users'
        user_data = {
            'name': '原始名称',
            'email': 'original@example.com',
            'password': 'test123'
        }
        create_response = requests.post(create_url, json=user_data, headers=auth_headers)
        user_id = create_response.json()['id']

        # 更新用户
        update_url = f'{BASE_URL}/users/{user_id}'
        update_data = {
            'name': '更新后的名称'
        }
        update_response = requests.put(update_url, json=update_data, headers=auth_headers)

        assert update_response.status_code == 200
        updated_user = update_response.json()
        assert updated_user['name'] == update_data['name']

    def test_delete_user(self, auth_headers):
        """测试删除用户"""
        # 先创建一个用户
        create_url = f'{BASE_URL}/users'
        user_data = {
            'name': '待删除用户',
            'email': 'delete@example.com',
            'password': 'test123'
        }
        create_response = requests.post(create_url, json=user_data, headers=auth_headers)
        user_id = create_response.json()['id']

        # 删除用户
        delete_url = f'{BASE_URL}/users/{user_id}'
        delete_response = requests.delete(delete_url, headers=auth_headers)

        assert delete_response.status_code == 200

        # 验证用户已被删除
        get_response = requests.get(delete_url, headers=auth_headers)
        assert get_response.status_code == 404

14.2 案例二：电商购物流程测试

模拟完整的电商购物流程：

import pytest
import requests

BASE_URL = 'https://api.example.com'

@pytest.fixture(scope="session")
def user_session():
    """用户会话"""
    session = requests.Session()
    session.base_url = BASE_URL

    # 登录
    login_response = session.post(f'{BASE_URL}/auth/login', json={
        'username': 'testuser',
        'password': 'test123'
    })
    token = login_response.json()['token']
    session.headers['Authorization'] = f'Bearer {token}'

    yield session

    # 登出
    session.post(f'{BASE_URL}/auth/logout')
    session.close()

@pytest.fixture(scope="function")
def cart(user_session):
    """购物车 fixture"""
    # 添加商品到购物车
    add_response = user_session.post(f'{BASE_URL}/cart/add', json={
        'product_id': 1,
        'quantity': 2
    })
    cart_id = add_response.json()['cart_id']

    yield cart_id

    # 清理：清空购物车
    try:
        user_session.delete(f'{BASE_URL}/cart/{cart_id}')
    except:
        pass

class TestShoppingFlow:
    """购物流程测试类"""

    def test_browse_products(self, user_session):
        """浏览商品"""
        response = user_session.get(f'{BASE_URL}/products')
        assert response.status_code == 200
        products = response.json()['products']
        assert len(products) > 0

    def test_add_to_cart(self, user_session):
        """添加到购物车"""
        response = user_session.post(f'{BASE_URL}/cart/add', json={
            'product_id': 1,
            'quantity': 2
        })
        assert response.status_code == 200
        assert 'cart_id' in response.json()

    def test_view_cart(self, user_session, cart):
        """查看购物车"""
        response = user_session.get(f'{BASE_URL}/cart/{cart}')
        assert response.status_code == 200
        cart_data = response.json()
        assert 'items' in cart_data
        assert len(cart_data['items']) > 0

    def test_create_order(self, user_session, cart):
        """创建订单"""
        response = user_session.post(f'{BASE_URL}/orders', json={
            'cart_id': cart,
            'shipping_address': '测试地址'
        })
        assert response.status_code == 201
        order = response.json()
        assert 'order_id' in order
        return order['order_id']

    def test_payment(self, user_session, cart):
        """支付订单"""
        # 先创建订单
        order_response = user_session.post(f'{BASE_URL}/orders', json={
            'cart_id': cart,
            'shipping_address': '测试地址'
        })
        order_id = order_response.json()['order_id']

        # 支付
        payment_response = user_session.post(f'{BASE_URL}/payments', json={
            'order_id': order_id,
            'payment_method': 'credit_card',
            'amount': 100.00
        })
        assert payment_response.status_code == 200
        payment = payment_response.json()
        assert payment['status'] == 'success'

14.3 案例三：参数化接口测试

使用 pytest 参数化测试多个接口：

import pytest
import requests

BASE_URL = 'https://api.example.com'

@pytest.fixture(scope="session")
def auth_headers():
    """认证请求头"""
    login_response = requests.post(f'{BASE_URL}/auth/login', json={
        'username': 'testuser',
        'password': 'test123'
    })
    token = login_response.json()['token']
    return {'Authorization': f'Bearer {token}'}

@pytest.mark.parametrize("endpoint,method,expected_status", [
    ('/users', 'GET', 200),
    ('/users/1', 'GET', 200),
    ('/products', 'GET', 200),
    ('/orders', 'GET', 200),
])
def test_api_endpoints(endpoint, method, expected_status, auth_headers):
    """参数化测试多个 API 端点"""
    url = f'{BASE_URL}{endpoint}'

    if method == 'GET':
        response = requests.get(url, headers=auth_headers)
    elif method == 'POST':
        response = requests.post(url, json={}, headers=auth_headers)

    assert response.status_code == expected_status, 
        f"端点 {endpoint} 返回了意外的状态码"

@pytest.mark.parametrize("username,password,expected_status", [
    ('admin', 'admin123', 200),
    ('testuser', 'test123', 200),
    ('wronguser', 'wrongpass', 401),
    ('', 'password', 400),
    ('username', '', 400),
])
def test_login_api(username, password, expected_status):
    """参数化测试登录接口"""
    url = f'{BASE_URL}/auth/login'
    data = {
        'username': username,
        'password': password
    }

    response = requests.post(url, json=data)
    assert response.status_code == expected_status, 
        f"用户名={username}, 密码={password} 的登录测试失败"

15. 最佳实践

15.1 使用 Fixture 管理资源

import pytest
import requests

@pytest.fixture(scope="session")
def api_session():
    """API Session Fixture"""
    session = requests.Session()
    session.base_url = 'https://api.example.com'

    # 登录
    login_response = session.post('/auth/login', json={
        'username': 'testuser',
        'password': 'test123'
    })
    token = login_response.json()['token']
    session.headers['Authorization'] = f'Bearer {token}'

    yield session

    # 清理
    session.post('/auth/logout')
    session.close()

15.2 统一错误处理

import pytest
import requests

def safe_request(method, url, **kwargs):
    """安全的请求函数，统一处理错误"""
    try:
        response = requests.request(method, url, **kwargs)
        response.raise_for_status()
        return response
    except requests.exceptions.Timeout:
        pytest.fail(f"请求超时：{url}")
    except requests.exceptions.ConnectionError:
        pytest.fail(f"连接失败：{url}")
    except requests.exceptions.HTTPError as e:
        pytest.fail(f"HTTP 错误：{e}")
    except requests.exceptions.RequestException as e:
        pytest.fail(f"请求异常：{e}")

def test_with_safe_request():
    """使用安全请求函数"""
    response = safe_request('GET', 'https://api.example.com/users')
    assert response.status_code == 200

15.3 响应数据验证

import pytest
import requests
from jsonschema import validate

# 定义 JSON Schema
USER_SCHEMA = {
    "type": "object",
    "properties": {
        "id": {"type": "integer"},
        "name": {"type": "string"},
        "email": {"type": "string", "format": "email"}
    },
    "required": ["id", "name", "email"]
}

def test_validate_response_schema():
    """验证响应数据格式"""
    response = requests.get('https://api.example.com/users/1')
    assert response.status_code == 200

    user_data = response.json()

    # 使用 JSON Schema 验证
    validate(instance=user_data, schema=USER_SCHEMA)

    # 手动验证
    assert isinstance(user_data['id'], int)
    assert isinstance(user_data['name'], str)
    assert '@' in user_data['email']

15.4 测试数据管理

import pytest
import requests
import json

@pytest.fixture
def test_data():
    """测试数据 Fixture"""
    return {
        'user': {
            'name': '测试用户',
            'email': 'test@example.com',
            'password': 'test123'
        },
        'product': {
            'name': '测试商品',
            'price': 99.99,
            'stock': 100
        }
    }

def test_create_user_with_test_data(test_data):
    """使用测试数据创建用户"""
    response = requests.post(
        'https://api.example.com/users',
        json=test_data['user']
    )
    assert response.status_code == 201

15.5 环境配置管理

import pytest
import requests
import os

@pytest.fixture(scope="session")
def api_base_url():
    """根据环境获取 API 基础 URL"""
    env = os.getenv('TEST_ENV', 'dev')

    urls = {
        'dev': 'https://dev-api.example.com',
        'test': 'https://test-api.example.com',
        'prod': 'https://api.example.com'
    }

    return urls.get(env, urls['dev'])

def test_with_env_config(api_base_url):
    """使用环境配置的测试"""
    response = requests.get(f'{api_base_url}/users')
    assert response.status_code == 200

16. 调试技巧

16.1 打印请求和响应信息

import pytest
import requests
import json

def test_debug_request():
    """调试请求和响应"""
    url = 'https://api.example.com/users'
    headers = {'Authorization': 'Bearer token123'}
    data = {'name': '测试用户'}

    # 打印请求信息
    print(f"请求 URL: {url}")
    print(f"请求头: {json.dumps(headers, indent=2, ensure_ascii=False)}")
    print(f"请求体: {json.dumps(data, indent=2, ensure_ascii=False)}")

    response = requests.post(url, json=data, headers=headers)

    # 打印响应信息
    print(f"状态码: {response.status_code}")
    print(f"响应头: {json.dumps(dict(response.headers), indent=2, ensure_ascii=False)}")
    print(f"响应体: {json.dumps(response.json(), indent=2, ensure_ascii=False)}")

    assert response.status_code == 201

16.2 使用 pytest-print 插件

import pytest
import requests

def test_with_print(print):
    """使用 pytest-print 打印信息"""
    response = requests.get('https://api.example.com/users')

    print(f"状态码: {response.status_code}")
    print(f"响应: {response.text}")

    assert response.status_code == 200

16.3 保存请求和响应到文件

import pytest
import requests
import json
from datetime import datetime

def test_save_request_response():
    """保存请求和响应到文件"""
    url = 'https://api.example.com/users'
    data = {'name': '测试用户'}

    response = requests.post(url, json=data)

    # 保存到文件
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    filename = f'request_response_{timestamp}.json'

    with open(filename, 'w', encoding='utf-8') as f:
        json.dump({
            'request': {
                'url': url,
                'method': 'POST',
                'data': data
            },
            'response': {
                'status_code': response.status_code,
                'headers': dict(response.headers),
                'body': response.json()
            }
        }, f, indent=2, ensure_ascii=False)

    assert response.status_code == 201

17. 性能测试

17.1 测量响应时间

import pytest
import requests
import time

def test_response_time():
    """测量响应时间"""
    url = 'https://api.example.com/users'

    start_time = time.time()
    response = requests.get(url)
    elapsed_time = time.time() - start_time

    assert response.status_code == 200
    assert elapsed_time < 1.0, f"响应时间过长：{elapsed_time:.2f} 秒"

    print(f"响应时间：{elapsed_time:.2f} 秒")

17.2 并发测试

import pytest
import requests
import concurrent.futures
import time

def test_concurrent_requests():
    """并发请求测试"""
    url = 'https://api.example.com/users'
    num_requests = 10

    def make_request():
        response = requests.get(url)
        return response.status_code

    start_time = time.time()

    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(make_request) for _ in range(num_requests)]
        results = [future.result() for future in concurrent.futures.as_completed(futures)]

    elapsed_time = time.time() - start_time

    assert all(status == 200 for status in results)
    print(f"并发 {num_requests} 个请求，耗时：{elapsed_time:.2f} 秒")

18. 常见问题和解决方案

18.1 中文编码问题

import pytest
import requests

def test_chinese_encoding():
    """处理中文编码"""
    url = 'https://api.example.com/users'
    data = {'name': '张三'}  # 包含中文

    # Requests 会自动处理编码
    response = requests.post(url, json=data)
    assert response.status_code == 201

    # 如果响应包含中文，确保使用正确的编码
    response.encoding = 'utf-8'
    print(response.text)

18.2 超时问题

import pytest
import requests

def test_handle_timeout():
    """处理超时问题"""
    url = 'https://api.example.com/slow-endpoint'

    try:
        # 设置合理的超时时间
        response = requests.get(url, timeout=5)
        assert response.status_code == 200
    except requests.exceptions.Timeout:
        # 超时处理
        pytest.skip("请求超时，跳过测试")

18.3 连接池问题

import pytest
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def test_connection_pool():
    """配置连接池"""
    session = requests.Session()

    # 配置重试和连接池
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504]
    )

    adapter = HTTPAdapter(
        max_retries=retry_strategy,
        pool_connections=10,
        pool_maxsize=20
    )

    session.mount("http://", adapter)
    session.mount("https://", adapter)

    response = session.get('https://api.example.com/users')
    assert response.status_code == 200

    session.close()

19. 总结

19.1 Requests 库的核心功能

  1. 简单易用：API 设计直观，学习成本低
  2. 功能全面：支持所有 HTTP 方法、认证、代理等
  3. 自动处理：自动处理编码、连接、重定向等
  4. 灵活配置：支持自定义请求头、超时、重试等

19.2 pytest 与 Requests 结合的优势

  1. 测试组织：使用 pytest 的 fixture 管理测试资源
  2. 参数化测试：轻松测试多个场景
  3. 断言清晰：使用 pytest 的断言机制
  4. 报告完善：生成详细的测试报告

19.3 学习建议

  1. 从基础开始：先掌握基本的 GET、POST 请求
  2. 实践为主：多写测试用例，积累经验
  3. 阅读文档：参考官方文档了解更多细节
  4. 代码复用：封装常用功能，提高效率

20. 附录

20.1 常用状态码

状态码 说明
200 请求成功
201 创建成功
204 无内容
400 请求错误
401 未授权
403 禁止访问
404 资源不存在
500 服务器错误
502 网关错误
503 服务不可用

20.2 常用请求头

请求头 说明 示例
Content-Type 内容类型 application/json
Authorization 认证信息 Bearer token123
User-Agent 用户代理 MyApp/1.0
Accept 接受的响应类型 application/json
X-API-Key API 密钥 your_api_key

20.3 参考资源

文档结束

发表评论