Channels
announcements
ask-marvin
best-practices-coordination-plane
data-ecosystem
data-tricks-and-tips
events
find-a-prefect-job
geo-bay-area
geo-berlin
geo-boston
geo-chicago
geo-colorado
geo-dc
geo-israel
geo-japan
geo-london
geo-nyc
geo-seattle
geo-texas
gratitude
introductions
marvin-in-the-wild
pacc-clearcover-june-12-2023
pacc-may-31-2023
ppcc-may-16-2023
prefect-ai
prefect-aws
prefect-azure
prefect-cloud
prefect-community
prefect-contributors
prefect-dbt
prefect-docker
prefect-gcp
prefect-getting-started
prefect-integrations
prefect-kubernetes
prefect-recipes
prefect-server
prefect-ui
random
show-us-what-you-got
Title
l
Luke Orland
12/06/2019, 4:51 PMI am developing a workflow, and I'm having trouble writing tests that mock out the
run()I've tried
unittest.mock.patchunittest.mock.patch.objectpytestfrom 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')mytask = SnowflakeQuery(**prefect.context.secrets.SNOWFLAKE).run(
query=myquery
)My high-level goal is to be able to develop the flow and its tasks while mocking out all I/O to the various databases and services that will be involved.
c
Chris White
12/06/2019, 5:17 PMHey @Luke Orland ! Because the Flow has already been created, and the task has already been instantiated, your mock isn’t having any effect - try mocking “my_task” and see if that solves it
l
Luke Orland
12/06/2019, 5:19 PMmy_task, in this case, is an instance of SnowflakeQuery, correct?
oh gotcha
I'm not seeing any difference.
Perhaps patching isn't the best way to develop workflows...
c
Chris White
12/06/2019, 5:30 PMActually, in the code snippet above it looks like you already called the run method of the task?
You should be patching the instance of the task that is associated to the flow, and it should work as you expect
l
Luke Orland
12/06/2019, 5:43 PMok, great, this is working:
from 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()💯 1
I removed the
.run(...)mytask(and move the
query=SnowflakeQuery().run(query=...)SnowflakeQuery(query=)c
Chris White
12/06/2019, 5:56 PMAwesome!!
@Marvin archive “How to mock a task run when testing a flow”
m
Marvin
12/06/2019, 5:58 PM