Add tests for log prefix and custom formatter in KubernetesPodOperator

- Add test_log_prefix_enabled to verify default log_prefix=True behavior
- Add test_log_prefix_disabled to confirm log_prefix=False removes container prefix
- Add test_custom_log_formatter to validate custom log_formatter functionality

Author: HsiuChuanHsu

kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py

@@ -1466,6 +1466,74 @@ def test_init_container_logs_filtered(self, mock_get_connection):
             < calls_args.find(marker_from_main_container)
         )
 
+    @pytest.mark.asyncio
+    def test_log_prefix_enabled(self, mock_get_connection, caplog):
+        """Test default behavior with log_prefix=True (container name prefix included)."""
+        marker = f"test_log_{uuid4()}"
+        k = KubernetesPodOperator(
+            namespace="default",
+            image="busybox",
+            cmds=["sh", "-cx"],
+            arguments=[f"echo {marker}"],
+            labels=self.labels,
+            task_id=str(uuid4()),
+            in_cluster=False,
+            do_xcom_push=False,
+            get_logs=True,
+            log_prefix=True,  # Explicitly set default
+        )
+        context = create_context(k)
+        with caplog.at_level(logging.INFO, logger="airflow.task.operators"):
+            k.execute(context)
+        assert any(f"[base] {marker}" in record.message for record in caplog.records)
+
+    @pytest.mark.asyncio
+    def test_log_prefix_disabled(self, mock_get_connection, caplog):
+        """Test log_prefix=False removes container name prefix."""
+        marker = f"test_log_{uuid4()}"
+        k = KubernetesPodOperator(
+            namespace="default",
+            image="busybox",
+            cmds=["sh", "-cx"],
+            arguments=[f"echo {marker}"],
+            labels=self.labels,
+            task_id=str(uuid4()),
+            in_cluster=False,
+            do_xcom_push=False,
+            get_logs=True,
+            log_prefix=False,
+        )
+        context = create_context(k)
+        with caplog.at_level(logging.INFO, logger="airflow.task.operators"):
+            k.execute(context)
+        assert any(marker in record.message and "[base]" not in record.message for record in caplog.records)
+
+    @pytest.mark.asyncio
+    def test_custom_log_formatter(self, mock_get_connection, caplog):
+        """Test custom log_formatter function."""
+        marker = f"test_log_{uuid4()}"
+
+        def custom_formatter(container_name: str, message: str) -> str:
+            return f"CUSTOM[{container_name}]: {message}"
+
+        k = KubernetesPodOperator(
+            namespace="default",
+            image="busybox",
+            cmds=["sh", "-cx"],
+            arguments=[f"echo {marker}"],
+            labels=self.labels,
+            task_id=str(uuid4()),
+            in_cluster=False,
+            do_xcom_push=False,
+            get_logs=True,
+            log_prefix=False,  # Ignored when log_formatter is provided
+            log_formatter=custom_formatter,
+        )
+        context = create_context(k)
+        with caplog.at_level(logging.INFO, logger="airflow.task.operators"):
+            k.execute(context)
+        assert any(f"CUSTOM[base]: {marker}" in record.message for record in caplog.records)
+
 
 # TODO: Task SDK: https://github.com/apache/airflow/issues/45438
 @pytest.mark.skip(reason="AIP-72: Secret Masking yet to be implemented")