Hi all!
I’m starting using prefect and I like it so far! 
I’m trying to write a test to verify that when something happens the error/issue is logged.
I tried to check here in the forum and on the web, but I did not found a solution/example.
Here the code:
# file: test_prefect_log.py
"""Test if log.error is call."""
from typing import Literal
from unittest.mock import MagicMock
import aiohttp
import pytest
from aioresponses import aioresponses
from prefect import get_run_logger
from prefect.logging import disable_run_logger
from prefect.testing.utilities import prefect_test_harness
from pytest_mock import MockerFixture
pytest_plugins = ("pytest_asyncio",)
# --------------------------- code to be tested -------------------------------
URL_STATIONS = "http://dati.retecivica.bz.it/services/meteo/v1/stations"
async def get_stations(
output_format: Literal["JSON", "CSV"] = "JSON",
coord_sys: str = "EPSG:4327",
) -> str:
"""Return a list with the meteo stations available in Alto Adige.
Available parameters are the string format and the geographical
coordination system.
"""
log = get_run_logger()
async with aiohttp.ClientSession() as session:
params = {"output_format": output_format, "coord_sys": coord_sys}
async with session.get(URL_STATIONS, params=params) as response:
log.info(f"Status: {response.status}")
log.info(f"Content-type: {response.headers['content-type']}")
if response.status != 200:
log.error("Return-code is different from 200! Check the code...")
return await response.text()
# -----------------------------------------------------------------------------
@pytest.fixture(autouse=True, scope="session")
def prefect_disable_logging() -> None:
"""Disable prefect logging error: prefect.exceptions.MissingContextError.
Without this fixture an exception is raised by the prefect logging system:
`MissingContextError: There is no active flow or task run context.`
"""
with disable_run_logger():
yield
@pytest.fixture(autouse=True, scope="session")
def prefect_test_fixture() -> None:
"""Use a dedicated temporary database for the tests."""
with prefect_test_harness():
yield
@pytest.mark.asyncio
async def test__get_stations_status404(mocker: MockerFixture) -> None:
"""Testing ddt4i.services.meteo.stations.get_stations function."""
spy_log_err = mocker.spy(MagicMock(), "error")
mocker.patch(
target="prefect.get_run_logger",
return_value=spy_log_err,
)
with aioresponses() as mocked:
mocked.get(
(
"http://dati.retecivica.bz.it/services/meteo/v1/"
"stations?coord_sys=EPSG%253A4327&output_format=JSON"
),
status=404,
body="",
)
await get_stations(output_format="JSON")
spy_log_err.assert_called_once()
The code assume that the following packages are installed on the venv:
pip install --upgrade aiohttp \
aioresponses \
pytest \
pytest-asyncio \
pytest_mock \
prefect
Now if I execute the code with:
» pytest test_prefect_log.py
==================================== test session starts ====================================
platform linux -- Python 3.10.10, pytest-7.2.2, pluggy-1.0.0
rootdir: /tmp/tst
plugins: asyncio-0.21.0, mock-3.10.0, anyio-3.6.2
asyncio: mode=strict
collected 1 item
test_prefect_log.py F [100%]
========================================= FAILURES ==========================================
_______________________________ test__get_stations_status404 ________________________________
__wrapped_mock_method__ = <function NonCallableMock.assert_called_once at 0x7f2b6f5a6b90>
args = (<MagicMock name='error' id='139824432632384'>,), kwargs = {}
__tracebackhide__ = True, msg = "Expected 'error' to have been called once. Called 0 times."
__mock_self = <MagicMock name='error' id='139824432632384'>
def assert_wrapper(
__wrapped_mock_method__: Callable[..., Any], *args: Any, **kwargs: Any
) -> None:
__tracebackhide__ = True
try:
> __wrapped_mock_method__(*args, **kwargs)
.venv/lib/python3.10/site-packages/pytest_mock/plugin.py:444:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <MagicMock name='error' id='139824432632384'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'error' to have been called once. Called 0 times.
/usr/lib/python3.10/unittest/mock.py:908: AssertionError
During handling of the above exception, another exception occurred:
mocker = <pytest_mock.plugin.MockerFixture object at 0x7f2b69bcf8e0>
@pytest.mark.asyncio
async def test__get_stations_status404(mocker: MockerFixture) -> None:
"""Testing ddt4i.services.meteo.stations.get_stations function."""
spy_log_err = mocker.spy(MagicMock(), "error")
mocker.patch(
target="prefect.get_run_logger",
return_value=spy_log_err,
)
with aioresponses() as mocked:
mocked.get(
(
"http://dati.retecivica.bz.it/services/meteo/v1/"
"stations?coord_sys=EPSG%253A4327&output_format=JSON"
),
status=404,
body="",
)
await get_stations(output_format="JSON")
> spy_log_err.assert_called_once()
E AssertionError: Expected 'error' to have been called once. Called 0 times.
test_prefect_log.py:79: AssertionError
----------------------------------- Captured stderr call ------------------------------------
16:06:24.629 | ERROR | null - Return-code is different from 200! Check the code...
------------------------------------- Captured log call -------------------------------------
ERROR null:test_prefect_log.py:35 Return-code is different from 200! Check the code...
================================== short test summary info ==================================
FAILED test_prefect_log.py::test__get_stations_status404 - AssertionError: Expected 'error' to have been called once. Called 0 times.
===================================== 1 failed in 1.03s =====================================
Do you know how to properly check that the log.error method is called?
