Remove global from listener unit tests

airflow-core/tests/unit/jobs/test_base_job.py

@@ -78,8 +78,8 @@ def test_base_job_respects_plugin_lifecycle(self, dag_maker):
         job_runner = MockJobRunner(job=job, func=lambda: sys.exit(0))
         run_job(job=job, execute_callable=job_runner._execute)
 
-        assert lifecycle_listener.started_component is job
-        assert lifecycle_listener.stopped_component is job
+        assert lifecycle_listener.get_listener_state().started_component is job
+        assert lifecycle_listener.get_listener_state().stopped_component is job
 
     def test_state_failed(self):
         def abort():

airflow-core/tests/unit/listeners/full_listener.py

@@ -17,45 +17,53 @@
 # under the License.
 from __future__ import annotations
 
+from dataclasses import dataclass, field
+from functools import cache
 from typing import Any
 
 from airflow.listeners import hookimpl
 from airflow.utils.state import TaskInstanceState
 
-started_component: Any = None
-stopped_component: Any = None
-state: list[Any] = []
+
+@dataclass
+class ListenerState:
+    started_component: Any = None
+    stopped_component: Any = None
+    state: list[Any] = field(default_factory=list)
+
+
+@cache
+def get_listener_state() -> ListenerState:
+    return ListenerState()
 
 
 @hookimpl
 def on_starting(component):
-    global started_component
-    started_component = component
+    get_listener_state().started_component = component
 
 
 @hookimpl
 def before_stopping(component):
-    global stopped_component
-    stopped_component = component
+    get_listener_state().stopped_component = component
 
 
 @hookimpl
 def on_task_instance_running(previous_state, task_instance):
-    state.append(TaskInstanceState.RUNNING)
+    get_listener_state().state.append(TaskInstanceState.RUNNING)
 
 
 @hookimpl
 def on_task_instance_success(previous_state, task_instance):
-    state.append(TaskInstanceState.SUCCESS)
+    get_listener_state().state.append(TaskInstanceState.SUCCESS)
 
 
 @hookimpl
 def on_task_instance_failed(previous_state, task_instance, error: None | str | BaseException):
-    state.append(TaskInstanceState.FAILED)
+    get_listener_state().state.append(TaskInstanceState.FAILED)
 
 
 def clear():
-    global started_component, stopped_component, state
-    started_component = None
-    stopped_component = None
-    state = []
+    listener_state = get_listener_state()
+    listener_state.started_component = None
+    listener_state.stopped_component = None
+    listener_state.state = []

airflow-core/tests/unit/listeners/lifecycle_listener.py

@@ -17,27 +17,35 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
+from functools import cache
 from typing import Any
 
 from airflow.listeners import hookimpl
 
-started_component: Any = None
-stopped_component: Any = None
+
+@dataclass
+class ListenerState:
+    started_component: Any = None
+    stopped_component: Any = None
+
+
+@cache
+def get_listener_state() -> ListenerState:
+    return ListenerState()
 
 
 @hookimpl
 def on_starting(component):
-    global started_component
-    started_component = component
+    get_listener_state().started_component = component
 
 
 @hookimpl
 def before_stopping(component):
-    global stopped_component
-    stopped_component = component
+    get_listener_state().stopped_component = component
 
 
 def clear():
-    global started_component, stopped_component
-    started_component = None
-    stopped_component = None
+    listener_state = get_listener_state()
+    listener_state.started_component = None
+    listener_state.stopped_component = None

airflow-core/tests/unit/listeners/test_listeners.py

@@ -79,8 +79,8 @@ def test_listener_gets_calls(create_task_instance, session):
     # `run()` calls `_run_raw_task()`
     ti.run()
 
-    assert len(full_listener.state) == 2
-    assert full_listener.state == [TaskInstanceState.RUNNING, TaskInstanceState.SUCCESS]
+    assert len(full_listener.get_listener_state().state) == 2
+    assert full_listener.get_listener_state().state == [TaskInstanceState.RUNNING, TaskInstanceState.SUCCESS]
 
 
 @provide_session
@@ -97,10 +97,10 @@ def test_multiple_listeners(create_task_instance, session):
         # suppress NotImplementedError: just for lifecycle
         run_job(job=job, execute_callable=job_runner._execute)
 
-    assert full_listener.started_component is job
-    assert lifecycle_listener.started_component is job
-    assert full_listener.stopped_component is job
-    assert lifecycle_listener.stopped_component is job
+    assert full_listener.get_listener_state().started_component is job
+    assert lifecycle_listener.get_listener_state().started_component is job
+    assert full_listener.get_listener_state().stopped_component is job
+    assert lifecycle_listener.get_listener_state().stopped_component is job
     assert class_based_listener.state == [DagRunState.RUNNING, DagRunState.SUCCESS]
 
 
@@ -140,8 +140,8 @@ def test_listener_captures_failed_taskinstances(create_task_instance_of_operator
     with pytest.raises(AirflowException):
         ti.run()
 
-    assert full_listener.state == [TaskInstanceState.RUNNING, TaskInstanceState.FAILED]
-    assert len(full_listener.state) == 2
+    assert full_listener.get_listener_state().state == [TaskInstanceState.RUNNING, TaskInstanceState.FAILED]
+    assert len(full_listener.get_listener_state().state) == 2
 
 
 @provide_session
@@ -154,8 +154,8 @@ def test_listener_captures_longrunning_taskinstances(create_task_instance_of_ope
     )
     ti.run()
 
-    assert full_listener.state == [TaskInstanceState.RUNNING, TaskInstanceState.SUCCESS]
-    assert len(full_listener.state) == 2
+    assert full_listener.get_listener_state().state == [TaskInstanceState.RUNNING, TaskInstanceState.SUCCESS]
+    assert len(full_listener.get_listener_state().state) == 2
 
 
 @provide_session