For our setup we re incorporating unittest in the build phas Prefect Community #ask-community

For our setup, we're incorporating unittest in the...

Nic

11/22/2022, 11:00 AM

For our setup, we're incorporating unittest in the build phase of our ci/cd pipeline. So that it won't deploy it to the cloud, unless the test passes in Azure devops I'd like to be able to run unittest on singular tasks and not start a flow run but when running

Copy code

import unittest
from tasks import address_matching as am
from prefect import flow,task


class TestDatahandling(unittest.TestCase):
    
    def test_tilslutning_forsyning(self):
        am.address_matching()



if __name__ == '__main__':
    unittest.main()

I get following error.message

Copy code

ERROR: test_tilslutning_forsyning (__main__.TestDatahandling)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "c:\Users\nho\Desktop\git\Prefect\etl\geoserver\tilslutning_forsyning\test.py", line 9, in test_tilslutning_forsyning
    am.address_matching()
  File "C:\Users\nho\AppData\Local\Programs\Python\Python310\lib\site-packages\prefect\tasks.py", line 353, in __call__
    return enter_task_run_engine(
  File "C:\Users\nho\AppData\Local\Programs\Python\Python310\lib\site-packages\prefect\engine.py", line 674, in enter_task_run_engine
    raise RuntimeError(
RuntimeError: Tasks cannot be run outside of a flow. To call the underlying task function outside of a flow use `task.fn()`.

Two questions 1. What would the syntax for task.fn() in my example be? I can't get it to run 2. Are there better ways, or best practices of running tests on flows or does this setup seems okay?

Khuyen Tran

11/22/2022, 4:54 PM

Have you tried

am.address_matching.fn()

Khuyen Tran

11/22/2022, 4:54 PM

I assume

am

is a class right?

Nic

11/22/2022, 5:10 PM

am is the imported module from the task file

Khuyen Tran

11/22/2022, 5:33 PM

and

am.address_matching

is a task?

Nic

11/22/2022, 5:51 PM

Yes, that's right

Nic

11/23/2022, 7:48 AM

@Khuyen Tran am, is the address_matching.py file, so it refers to the address_matching task in the file of the same name. In my flow it's called like so

Copy code

@flow()
def tilslutning_forsyning():
    tilslutning_matched, forsyning_matched = am.address_matching()

    am.write_to_database(tilslutning_matched, "connection_obligation")

    am.write_to_database(forsyning_matched, "heating_supply_network")

However, i aim to test the tasks before deploy, wanting to call singular tasks in the test file, but without running a flow so i don't clock up the ui with test runs in prefect.io

Nic

11/23/2022, 7:59 AM

It works with your code, plus disabling the logger

Copy code

import unittest
from tasks import address_matching as am
from prefect import task
from prefect.logging import disable_run_logger


class TestDatahandling(unittest.TestCase):
    
    def test_tilslutning_forsyning(self):
        with disable_run_logger():
            am.address_matching.fn()



if __name__ == '__main__':
    unittest.main()

🎉 1

4 Views

Open in Slack

Previous Next