Luke Orland
12/06/2019, 4:51 PMrun()
method of tasks in the workflow.unittest.mock.patch
, unittest.mock.patch.object
, also followed the examples in the prefect codebase with `monkeypatch.
With the following test (run with pytest
), I'm finding that the query is being executed for real and not patched.
from unittest.mock import MagicMock
from myworkflow import flow
def test_flow(monkeypatch):
MockSnowflakeQuery = MagicMock()
mock_run = MagicMock()
MockSnowflakeQuery.run = mock_run
monkeypatch.setattr(
"myworkflow.SnowflakeQuery", MockSnowflakeQuery,
)
flow.run()
args, kwargs = mock_run.call_args
assert args == ()
assert kwargs['query'].startswith('COPY INTO')
The flow has a task:
mytask = SnowflakeQuery(**prefect.context.secrets.SNOWFLAKE).run(
query=myquery
)
Chris White
12/06/2019, 5:17 PMLuke Orland
12/06/2019, 5:19 PMChris White
12/06/2019, 5:30 PMLuke Orland
12/06/2019, 5:43 PMfrom unittest.mock import patch
from kepler.recommendations.workflow import flow, mytask
@patch.object(mytask, 'run')
def test_flow(mock_mytask_run):
flow.run()
mock_mytask_run.assert_called()
Thanks!.run(...)
call from the SnowflakeQuery() instance in the assignment of mytask
query=
kwarg from SnowflakeQuery().run(query=...)
into SnowflakeQuery(query=)
)Chris White
12/06/2019, 5:56 PMMarvin
12/06/2019, 5:58 PM