davzucky
11/24/2022, 11:51 PMget_run_logger()
which is set from the context. You can find sample test code on the thread
The test keep failing with the erorr
prefect.exceptions.MissingContextError: There is no active flow or task run context.
import pytest
from prefect import flow, task, get_run_logger
from prefect.testing.utilities import prefect_test_harness
@pytest.fixture(autouse=True, scope="session")
def prefect_test_fixture():
with prefect_test_harness():
yield
@task
def sum(a: int, b: int) -> int:
return a + b
@task
def sum_with_logger(a: int, b: int) -> int:
logger = get_run_logger()
logger.debug(f"sum {a}+{b}={a+b}")
return a + b
@flow
def sum_flow_with_logger(a: int, b: int) -> int:
return sum_with_logger(a, b)
def test_sum():
# Test pass
a = 1
b = 3
result = sum.fn(a, b)
assert 4 == result
def test_sum_with_logger():
# test error with prefect.exceptions.MissingContextError: There is no active flow or task run context.
a = 1
b = 3
result = sum_with_logger.fn(a, b)
assert 4 == result
def test_sum_flow_with_logger():
# test error with prefect.exceptions.MissingContextError: There is no active flow or task run context.
a = 1
b = 3
result = sum_flow_with_logger.fn(a, b)
assert 4 == result
Peyton Runyan
11/25/2022, 12:18 AMex.py
which has our example flow
from prefect import flow, get_run_logger
@flow
def my_flow():
get_run_logger().info("I am logging stuff")
return 42
and we have our test file test_ex.py
I would probably do something similar to the following:
from unittest.mock import Mock
from ex import my_flow
def test_my_flow(monkeypatch):
mock_logger = Mock()
monkeypatch.setattr("ex.get_run_logger", mock_logger)
res = my_flow()
...
from unittest.mock import Mock
from ex import my_flow
import pytest
@pytest.fixture
def mock_logger(monkeypatch):
mock_logger = Mock()
monkeypatch.setattr("ex.get_run_logger", mock_logger)
return mock_logger
def test_my_flow(mock_logger):
res = my_flow()
davzucky
11/25/2022, 12:24 AMPeyton Runyan
11/25/2022, 12:26 AMKrishnan Chandra
11/25/2022, 12:27 AMfrom prefect.logging import disable_run_logger
from prefect.testing.utilities import prefect_test_harness
@pytest.fixture(autouse=True, scope="session")
def prefect_disable_logging():
with disable_run_logger():
yield
@pytest.fixture(autouse=True, scope="session")
def prefect_test_fixture():
with prefect_test_harness():
yield
Peyton Runyan
11/25/2022, 12:32 AMdavzucky
11/25/2022, 12:32 AMdisable_run_logger
not be in
prefect.testing.utilities
be a better place? as I don't know when you would use it outside of testingAnna Geller
11/25/2022, 12:49 AM