Fix KubernetesExecutor test

    - Previous test failure is cuased by cache state of executor_instances
    - Should set ti.state = RUNNING after ti.run

Author: jason810496

providers/tests/cncf/kubernetes/log_handlers/test_log_handlers.py

@@ -43,6 +43,7 @@
 
 from tests_common.test_utils.compat import PythonOperator
 from tests_common.test_utils.config import conf_vars
+from tests_common.test_utils.executor_loader import clean_executor_loader
 from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS
 
 if AIRFLOW_V_3_0_PLUS:
@@ -76,6 +77,7 @@ def teardown_method(self):
     @pytest.mark.parametrize("state", [TaskInstanceState.RUNNING, TaskInstanceState.SUCCESS])
     def test__read_for_k8s_executor(self, mock_k8s_get_task_log, create_task_instance, state):
         """Test for k8s executor, the log is read from get_task_log method"""
+        clean_executor_loader()
         mock_k8s_get_task_log.return_value = ([], [])
         executor_name = "KubernetesExecutor"
         ti = create_task_instance(
@@ -86,6 +88,7 @@ def test__read_for_k8s_executor(self, mock_k8s_get_task_log, create_task_instanc
         )
         ti.state = state
         ti.triggerer_job = None
+        ti.executor = executor_name
         with conf_vars({("core", "executor"): executor_name}):
             reload(executor_loader)
             fth = FileTaskHandler("")
@@ -105,11 +108,12 @@ def test__read_for_k8s_executor(self, mock_k8s_get_task_log, create_task_instanc
             pytest.param(k8s.V1Pod(metadata=k8s.V1ObjectMeta(name="pod-name-xxx")), "default"),
         ],
     )
-    @patch.dict("os.environ", AIRFLOW__CORE__EXECUTOR="KubernetesExecutor")
+    @conf_vars({("core", "executor"): "KubernetesExecutor"})
     @patch("airflow.providers.cncf.kubernetes.kube_client.get_kube_client")
     def test_read_from_k8s_under_multi_namespace_mode(
         self, mock_kube_client, pod_override, namespace_to_call
     ):
+        reload(executor_loader)
         mock_read_log = mock_kube_client.return_value.read_namespaced_pod_log
         mock_list_pod = mock_kube_client.return_value.list_namespaced_pod
 
@@ -139,6 +143,7 @@ def task_callable(ti):
         )
         ti = TaskInstance(task=task, run_id=dagrun.run_id)
         ti.try_number = 3
+        ti.executor = "KubernetesExecutor"
 
         logger = ti.log
         ti.log.disabled = False
@@ -147,6 +152,8 @@ def task_callable(ti):
         set_context(logger, ti)
         ti.run(ignore_ti_state=True)
         ti.state = TaskInstanceState.RUNNING
+        # clear executor_instances cache
+        file_handler.executor_instances = {}
         file_handler.read(ti, 2)
 
         # first we find pod name

tests/utils/test_log_handlers.py

@@ -238,7 +238,6 @@ def test_file_task_handler_with_multiple_executors(
             )
             if executor_name is not None:
                 ti.executor = executor_name
-            ti.state = TaskInstanceState.RUNNING
             ti.try_number = 1
             logger = ti.log
             ti.log.disabled = False
@@ -249,13 +248,16 @@ def test_file_task_handler_with_multiple_executors(
             assert file_handler is not None
 
             set_context(logger, ti)
+            # clear executor_instances cache
+            file_handler.executor_instances = {}
             assert file_handler.handler is not None
             # We expect set_context generates a file locally.
             log_filename = file_handler.handler.baseFilename
             assert os.path.isfile(log_filename)
             assert log_filename.endswith("1.log"), log_filename
 
             ti.run(ignore_ti_state=True)
+            ti.state = TaskInstanceState.RUNNING
 
             file_handler.flush()
             file_handler.close()