I have an api with a route that creates google cloud tasks. I import the task client as dependency into the endpoint. I mocked the task client in my tests with app.dependency_overrides
but now I’m not sure how to use that in order to see the call args for the send_task_to_queue
function. In my old flask api I could just patch send_task_to_queue
and check its call_args
but trying that here is giving me None
<code># TASK CLIENT
class TasksClient:
def __init__(self) -> None:
# sets up client
def send_task_to_queue(
self,
queue: str,
task_func: str,
payload: Any # payload can be anything
) -> None:
# creates task - want to check the queue, task_func, and payload args
@lru_cache
def get_tasks_client() -> TasksClient:
return TasksClient()
# ENDPOINT
@router.post("/route",
status_code=201,
description="POST to route")
async def post_inspection(
req: Request,
tasks_client: TasksClient = Depends(get_tasks_client)
) -> None:
tasks_client.send_task_to_queue('queue_name', 'task_func_name', 'payload')
# MOCK CLASSES
class MockGoogleTaskClient(object):
""" Mock tasks client from google """
@staticmethod
def queue_path(project_id, loc, queue):
return f'{project_id}.{loc}.{queue}'
@staticmethod
def create_task(path, task):
return
class MockTasksClient(object):
""" mock tasks class """
def __init__(self):
self.credentials = 'creds'
self.client = MockGoogleTaskClient()
@staticmethod
def send_task_to_queue(queue, task_func, payload):
""" mock send task to queue """
return
def mock_get_tasks_client() -> MockTasksClient:
return MockTasksClient()
# CONFTEST
from api.main import app
app.dependency_overrides[get_tasks_client] = mock_get_tasks_client
# TEST
# doesn't work - get None
@patch('evir_api_v2.util.tasks.get_tasks_client', side_effect=mock_get_tasks_client)
def test_post_inspection(
m_tasks_client,
client: TestClient,
engine: Engine
) -> None:
happy_insert_resp = client.post(f'route', json={'hello': 'test'})
assert happy_insert_resp.status_code == 201
# make sure tasks are created
assert m_tasks_client.send_task_to_queue.call_args == ??? # comes in as None
</code>
<code># TASK CLIENT
class TasksClient:
def __init__(self) -> None:
# sets up client
def send_task_to_queue(
self,
queue: str,
task_func: str,
payload: Any # payload can be anything
) -> None:
# creates task - want to check the queue, task_func, and payload args
@lru_cache
def get_tasks_client() -> TasksClient:
return TasksClient()
# ENDPOINT
@router.post("/route",
status_code=201,
description="POST to route")
async def post_inspection(
req: Request,
tasks_client: TasksClient = Depends(get_tasks_client)
) -> None:
tasks_client.send_task_to_queue('queue_name', 'task_func_name', 'payload')
# MOCK CLASSES
class MockGoogleTaskClient(object):
""" Mock tasks client from google """
@staticmethod
def queue_path(project_id, loc, queue):
return f'{project_id}.{loc}.{queue}'
@staticmethod
def create_task(path, task):
return
class MockTasksClient(object):
""" mock tasks class """
def __init__(self):
self.credentials = 'creds'
self.client = MockGoogleTaskClient()
@staticmethod
def send_task_to_queue(queue, task_func, payload):
""" mock send task to queue """
return
def mock_get_tasks_client() -> MockTasksClient:
return MockTasksClient()
# CONFTEST
from api.main import app
app.dependency_overrides[get_tasks_client] = mock_get_tasks_client
# TEST
# doesn't work - get None
@patch('evir_api_v2.util.tasks.get_tasks_client', side_effect=mock_get_tasks_client)
def test_post_inspection(
m_tasks_client,
client: TestClient,
engine: Engine
) -> None:
happy_insert_resp = client.post(f'route', json={'hello': 'test'})
assert happy_insert_resp.status_code == 201
# make sure tasks are created
assert m_tasks_client.send_task_to_queue.call_args == ??? # comes in as None
</code>
# TASK CLIENT
class TasksClient:
def __init__(self) -> None:
# sets up client
def send_task_to_queue(
self,
queue: str,
task_func: str,
payload: Any # payload can be anything
) -> None:
# creates task - want to check the queue, task_func, and payload args
@lru_cache
def get_tasks_client() -> TasksClient:
return TasksClient()
# ENDPOINT
@router.post("/route",
status_code=201,
description="POST to route")
async def post_inspection(
req: Request,
tasks_client: TasksClient = Depends(get_tasks_client)
) -> None:
tasks_client.send_task_to_queue('queue_name', 'task_func_name', 'payload')
# MOCK CLASSES
class MockGoogleTaskClient(object):
""" Mock tasks client from google """
@staticmethod
def queue_path(project_id, loc, queue):
return f'{project_id}.{loc}.{queue}'
@staticmethod
def create_task(path, task):
return
class MockTasksClient(object):
""" mock tasks class """
def __init__(self):
self.credentials = 'creds'
self.client = MockGoogleTaskClient()
@staticmethod
def send_task_to_queue(queue, task_func, payload):
""" mock send task to queue """
return
def mock_get_tasks_client() -> MockTasksClient:
return MockTasksClient()
# CONFTEST
from api.main import app
app.dependency_overrides[get_tasks_client] = mock_get_tasks_client
# TEST
# doesn't work - get None
@patch('evir_api_v2.util.tasks.get_tasks_client', side_effect=mock_get_tasks_client)
def test_post_inspection(
m_tasks_client,
client: TestClient,
engine: Engine
) -> None:
happy_insert_resp = client.post(f'route', json={'hello': 'test'})
assert happy_insert_resp.status_code == 201
# make sure tasks are created
assert m_tasks_client.send_task_to_queue.call_args == ??? # comes in as None