python - 如何使用 pytest aiohttp 在 python 中运行异步 http 测试
问题描述
我想测试对我不拥有的 api 的 http 调用(集成测试)。我创建了一个带有异步函数的类,该函数使用 aiohttp 进行 http 调用。然后我做了一个测试用例来运行这个函数并断言它的状态。
测试/test_coinbase.py:
import json
import os
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
from src.coinbase import CoinbaseService
class CoinbaseTestCase():
async def test_get_current_user(self):
coinbase_service = CoinbaseService(os.getenv('COINBASE_APIURL'), os.getenv('COINBASE_APIKEY'), os.getenv('COINBASE_APISECRET'))
status, user = await coinbase_service.show_current_user()
assert status == 200
src/coinbase.py:
import json, hmac, hashlib, time
import aiohttp
class CoinbaseService:
def __init__(self, API_URL, API_KEY, API_SECRET):
self.API_URL = API_URL
self.API_KEY = API_KEY
self.API_SECRET = API_SECRET
def generateHeaders(self, method, path_url, body = ''):
timestamp = str(int(time.time()))
message = timestamp + method + path_url + body
signature = hmac.new(self.API_SECRET, message, hashlib.sha256).hexdigest()
headers = {
'CB-ACCESS-SIGN': signature,
'CB-ACCESS-TIMESTAMP': timestamp,
'CB-ACCESS-KEY': self.API_KEY
}
return headers
async def show_current_user(self):
path_url = 'user'
headers = self.generateHeaders('GET', path_url)
async with aiohttp.ClientSession() as session:
async with session.get(path_url, headers=headers) as response:
status = response.status
user = await response.json()
return status, user
当我在我的根项目中运行以下命令时,我得到以下信息。
平台 darwin -- Python 3.7.4, pytest-5.3.5, py-1.8.1, pluggy-0.13.1 rootdir: /Users/helloworld/Krypto plugins: aiohttp-0.3.0 收集 0 个项目
套餐
[packages]
aiohttp = "*"
backoff = "*"
requests = "*"
asyncio = "*"
vadersentiment = "*"
python-dateutil = "*"
pytest = "*"
pytest-aiohttp = "*"
解决方案
CoinbaseTestCase
与默认命名规则不匹配pytest
被视为测试类,请参阅Python 测试发现的约定。要么将类重命名为类似的名称TestCoinbase
,要么定义您自己的命名约定。vanilla
pytest
不支持运行异步测试。安装pytest-asyncio
:$ pip install pytest-asyncio
并用标记标记异步测试
asyncio
:class TestCoinbase: @pytest.mark.asyncio async def test_get_current_user(self): await coinbase_service.show_current_user() ...
推荐阅读
- angular - 如何将此表示日期的字符串转换为正确的 FireStore 时间戳字段?
- java - java-给处理中的数组赋值
- python - 向 seaborn 条添加标签
- sql - 如何根据父列值为所有子记录(包括父记录)填充事实列(屏幕截图)
- reactjs - 角度组件和 React 组件有什么区别?
- r - 在 Docker 容器上运行管道工 API
- swift - 错误:无法构建 Objective-C 模块“快速”
- unicode - NCLOB 到 Unicode/HEX/BASE64 的转换
- database-design - 建模数据库中实体的逻辑表达式
- xcode - 无法在 macOS 上使用 XCode 项目配置 OpenCv 和 PCL