问题描述
我想使用 pytest 和 mock 对以下 Flask API 进行单元测试:
#api.py
@api.route("/get_job/<job_name>")
class JobStatus(Resource):
@api.marshal_with(m_res)
def get(self,job_name):
try:
job_status = gcp.generate_signed_url("bucket_name",blob_path)
except ValueError as e:
try:
client = kube.SparkApplicationClient()
job_status = client.get(job_name)["status"]["applicationState"]["state"]
except ApiException as e:
abort(404,"job doesn't exist: %s\n" % e)
return {"job_status": job_status}
我针对发送了两个异常但补丁不起作用的情况编写了以下测试:
#test_api.py
import pytest
from unittest.mock import patch,Magicmock
from kubernetes.client.rest import ApiException
from api import server
@pytest.fixture
def app():
app = server.app
return app
@patch("api.gcp.generate_signed_url",Magicmock(side_effect=Exception(ValueError)))
@patch("api.kube.SparkApplicationClient.get",Magicmock(side_effect=Exception(ApiException)))
def test_get_job_status_doesnt_exists(client):
response = client.get(
"/api/get_job/68987af5-ccf4-4c11-a89a-d6645b7f7"
)
assert response.status_code == 404
基本上我试图让 gcp.generate_signed_url
和 client.get(job_name)
在测试时引发异常
谢谢!!! :)
解决方法
试试:
@patch(
"api.gcp.generate_signed_url",side_effect=ValueError)
)
还有:
@patch(
"api.kube.SparkApplicationClient.get",**{
'return_value.get.side_effect': ApiException
}
)