Nic
11/22/2022, 11:00 AMimport 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
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 PMam.address_matching.fn()
?am
is a class right?Nic
11/22/2022, 5:10 PMKhuyen Tran
11/22/2022, 5:33 PMam.address_matching
is a task?Nic
11/22/2022, 5:51 PM@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.ioimport 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()