GitHub user DonHaul created a discussion: How do you test tasks that use
connections?
Hey all, so our team was using airflow `3.0.3` and currently upgraded to
`3.1.4`. We use `pytest` to test our tasks individually and this last upgrade
broke them, mostly because connections are no longer acessible outside
airflow's executions time.
say this test
```python
class Test_HEPCreateDAG:
dag = dagbag.get_dag("hep_create_dag")
context = {
"dag_run": {"run_id": "test_run"},
"ti": {"xcom_push": lambda key, value: None},
"params": {"workflow_id": "00000000-0000-0000-0000-000000001111"},
}
def test_set_schema(self):
task = self.dag.get_task("set_schema")
result = task.execute(context={})
```
where this set_schema task does something like:
```python
from airflow.sdk.bases.hook import BaseHook
@task
def set_schema(**context):
conn = BaseHook.get_connection("my_connection")
```
would work but now it fails with
```
airflow.exceptions.AirflowNotFoundException: The conn_id `my_connection` isn't
defined
```
So I am wondering, how do you guys test tasks that require accessing
connections?
GitHub link: https://github.com/apache/airflow/discussions/59364
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
