Unit testing @task decorated functions without active Flow context.

[skwzrd]

Hello,

I am trying to run my unittests after decorating some functions with @task and now I am getting the error below because the tasks are not associated with any Flows. Can I override this somehow and tell Prefect to never mind the @task decorators? I just want to run tests against my functions, not the Task capabilities.

ERROR: test_transform_data (__main__.Tests)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "C:\Users\skwzrd\AppData\Local\Programs\Python\Python38\lib\unittest\mock.py", line 1325, in patched
    return func(*newargs, **newkeywargs)
  File "test.py", line 115, in test_transform_data
    main.transform_data(raw_data)
  File "C:\Users\skwzrd\AppData\Local\Programs\Python\Python38\lib\site-packages\prefect\core\task.py", line 582, in __call__
    new.bind(
  File "C:\Users\skwzrd\AppData\Local\Programs\Python\Python38\lib\site-packages\prefect\core\task.py", line 635, in bind
    raise ValueError("Could not infer an active Flow context.")
ValueError: Could not infer an active Flow context.

Cheers,
skwzrd

[zanieb]

Hi @skwzrd -- the typical pattern here is to call my_task.run() in your tests, for example

@task
def foo(arg1, arg2):
    return arg1 + arg2

def test_foo():
   assert foo.run(1, 2) == 3

[skwzrd]

thanks a bunch @madkinsz !

[bennnym]

Hi there,

I have ran into an issue that I am simplifying with a more straight forward example.

I have all my tasks as functions in one file, with a Flow as a context manager.

I have another file where I am running my tests with pytest.

The default behaviour of pytest seems to be that the entire file is evaluated and as a result I am not able to isolate my functions logic for testing.

flow.py

from prefect import task, Flow
class Storage:
    @staticmethod
    def bar():
        print("Strage.bar CALLED FOR SOME REASON??")
@task
def foo_task():
    print("foo_task")
with Flow(
    name="example-flow",
    storage=Storage.bar()
) as flow:
    foo = foo_task()

test.py

   from the_flow import foo_task
class TestFlow:
    def test_foo_task(self):
        foo_task.run()
        assert 1==2

When running pytest -s flow.py

I see the output of "Strage.bar CALLED FOR SOME REASON??"

Care to explain if there is a way around this?

I have tried to mock Flow but that doesnt work either..

[mbopfNIH]

@skwzrd Did you ever find a solution to this? I am experiencing the same problem.