Pooyan
12/19/2023, 9:05 PMrun_count
in unit tests? or another way of putting it is how can I have access to retry_delay_seconds
and retries
when running a flow?Marvin
12/19/2023, 9:06 PMrun_count
of a flow or accessing retry_delay_seconds
and retries
in Prefect unit tests can be done programmatically, but the query did not return detailed instructions or specific code examples relevant to your question. However, based on common testing patterns in Python, I can provide some general guidance.
To mock or access these values during a flow run, you would typically use mocking libraries such as unittest.mock
in Python. Here's how you might approach this:
1. Mocking `run_count`:
You could use a mocking library to replace the run_count
value with a mocked one. This involves patching the object that tracks the run_count
within Prefect's code so that when your test runs, it uses the mocked value instead.
2. Accessing retry_delay_seconds
and `retries`:
During a flow run, these parameters can be specified as part of the flow or task configuration. When writing tests, you can manually set these parameters in the flow definition and inspect them as needed.
Here's an outline of what the code in your unit tests might look like:
python
from unittest.mock import patch
from prefect import Flow
import your_flow_module # This should be the module where your flow is defined
# Mocking run_count example
with patch("your_flow_module.some_object.run_count", new_callable=PropertyMock) as mock_run_count:
mock_run_count.return_value = 3 # Assuming you want to simulate 3 runs
# Proceed to run your flow and tests here
# Setting and accessing retry_delay_seconds and retries example
with Flow("test-flow", retries=2, retry_delay_seconds=10) as flow:
# Define tasks and flow structure here
# Now you can run your flow and check the retries and retry_delay_seconds
flow.run()
assert flow.retries == 2
assert flow.retry_delay_seconds == 10
For a more specific answer, I recommend checking the documentation or Prefect's testing guidelines to see if there are updated or recommended approaches for testing these scenarios in Prefect 2.x. Alternatively, I can perform a GitHub issue search in the PrefectHQ/prefect repository if you would like to see if there's any previously reported issue or discussion related to this topic. Would you like me to do so?Nate
12/19/2023, 9:09 PMNate
12/19/2023, 9:10 PMIn [1]: from prefect import flow
In [2]: @flow(log_prints=True, retries=1)
...: def foo():
...: print(foo.retries, foo.retry_delay_seconds)
...:
In [3]: foo()
15:10:08.443 | INFO | prefect.engine - Created flow run 'vivacious-seal' for flow 'foo'
15:10:08.899 | INFO | Flow run 'vivacious-seal' - 1 0
15:10:09.869 | INFO | Flow run 'vivacious-seal' - Finished in state Completed()
Pooyan
12/19/2023, 9:20 PMwith Flow
solution? Can I just override the original retry and retry_delay_seconds there when doing unit tests?Nate
12/19/2023, 9:21 PMNate
12/19/2023, 9:22 PMNate
12/19/2023, 9:22 PMmy_flow.with_options(retries=42)(*args, **kwargs)
in my testPooyan
12/19/2023, 9:23 PMPooyan
12/19/2023, 9:23 PM