Channels
#prefect-community
Title
n
Nic
11/22/2022, 11:00 AMFor 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() 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()`.k
Khuyen Tran
11/22/2022, 4:54 PMHave you tried
am.address_matching.fn()I assume
amn
Nic
11/22/2022, 5:10 PMam is the imported module from the task file
k
Khuyen Tran
11/22/2022, 5:33 PMand
am.address_matchingn
Nic
11/22/2022, 5:51 PMYes, 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")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
3 Views