For our setup, we're incorporating unittest in the...
# prefect-community
n
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?
k
Have you tried
am.address_matching.fn()
?
I assume
am
is a class right?
n
am is the imported module from the task file
k
and
am.address_matching
is a task?
n
Yes, that's right
@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
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