Is there a way to unit test tasks where `.fn` retu...
# prefect-community
m
Is there a way to unit test tasks where
.fn
returns a coroutine? I tried this:
from prefect_aws.s3 import s3_upload
...
with prefect_test_harness():
my_upload = s3_upload.fn(bucket=S3_BUCKET_NAME, key='B5_key', data=b'55555', aws_credentials = AwsCredentials())
asyncio.run(my_upload)
But it’s not working with this runtime error:
Copy code
E           RuntimeError: There is no active flow or task run context.

../valo-prefect-poc/.venv/lib/python3.7/site-packages/prefect/logging/loggers.py:91: RuntimeError
discourse 1
Probably not important since I can always wrap in a
@flow
- mostly just curious. 😛
k
Yeah I’ve been looking and it seems that the task library tests just wrap in a test flow like this. I don’t think this can be done
But I’ll also ask some team members to be sure. I’ll let you know when I get a response
gratitude thank you 1
z
Hm… you should be able to test the underlying function still
Ah including the full traceback is important
Copy code
Traceback (most recent call last):
  File "/Users/mz/flow.py", line 30, in <module>
    asyncio.run(my_upload)
  File "/opt/homebrew/Caskroom/miniconda/base/envs/orion-dev-38/lib/python3.8/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/opt/homebrew/Caskroom/miniconda/base/envs/orion-dev-38/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "/opt/homebrew/Caskroom/miniconda/base/envs/orion-dev-38/lib/python3.8/site-packages/prefect_aws/s3.py", line 126, in s3_upload
    logger = get_run_logger()
  File "/Users/mz/dev/orion/src/prefect/logging/loggers.py", line 91, in get_run_logger
    raise RuntimeError("There is no active flow or task run context.")
RuntimeError: There is no active flow or task run context.
This is because the task uses a logger
We’ll likely find a way to allow run loggers to be retrieved during testing without throwing an exception
👍 1
cc @alex just so this is on your radar
👍 1