Fix 3.0 compact

Author: jason810496

devel-common/src/tests_common/test_utils/version_compat.py

@@ -34,4 +34,4 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
 
 AIRFLOW_V_2_10_PLUS = get_base_airflow_version_tuple() >= (2, 10, 0)
 AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)
-[].sort()
+AIRFLOW_V_3_0 = get_base_airflow_version_tuple() == (3, 0, 0)

providers/celery/tests/unit/celery/log_handlers/test_log_handlers.py

@@ -40,7 +40,7 @@
 from tests_common.test_utils.file_task_handler import (
     convert_list_to_stream,
 )
-from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_0, AIRFLOW_V_3_0_PLUS
 
 pytestmark = pytest.mark.db_test
 
@@ -95,10 +95,13 @@ def test__read_for_celery_executor_fallbacks_to_worker(self, create_task_instanc
             fth._read_from_logs_server.assert_called_once()
 
         if AIRFLOW_V_3_0_PLUS:
-            assert metadata == {"end_of_log": False, "first_time_read": False}
             logs = list(logs)
             assert logs[0].sources == ["this message"]
             assert [x.event for x in logs[-3:]] == ["this", "log", "content"]
+            if AIRFLOW_V_3_0:
+                assert metadata == {"end_of_log": False, "first_time_read": False}
+            else:
+                assert metadata == {"end_of_log": False, "log_pos": 3}
         else:
             assert "*** this message\n" in logs
             assert logs.endswith("this\nlog\ncontent")

providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py

@@ -44,7 +44,10 @@
     ElasticsearchJSONFormatter,
 )
 from airflow.providers.elasticsearch.log.es_response import ElasticSearchResponse, Hit
-from airflow.providers.elasticsearch.version_compat import AIRFLOW_V_3_0_PLUS
+from airflow.providers.elasticsearch.version_compat import (
+    AIRFLOW_V_3_0,
+    AIRFLOW_V_3_0_PLUS,
+)
 from airflow.utils import timezone
 from airflow.utils.log.file_task_handler import FileTaskHandler
 from airflow.utils.log.logging_mixin import ExternalLoggingMixin, LoggingMixin
@@ -358,7 +361,7 @@ def _read(
                     "If your task started recently, please wait a moment and reload this page. "
                     "Otherwise, the logs for this task instance may have been removed."
                 )
-                if AIRFLOW_V_3_0_PLUS:
+                if AIRFLOW_V_3_0_PLUS and not AIRFLOW_V_3_0:
                     from airflow.utils.log.file_task_handler import (
                         StructuredLogMessage,
                         get_compatible_output_log_stream,
@@ -390,7 +393,7 @@ def concat_logs(hits: list[Hit]) -> str:
             return "\n".join(self._format_msg(hits[i]) for i in range(log_range))
 
         if logs_by_host:
-            if AIRFLOW_V_3_0_PLUS:
+            if AIRFLOW_V_3_0_PLUS and not AIRFLOW_V_3_0:
                 from airflow.utils.log.file_task_handler import (
                     StructuredLogMessage,
                     get_compatible_output_log_stream,

providers/elasticsearch/src/airflow/providers/elasticsearch/version_compat.py

@@ -34,3 +34,4 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
 
 AIRFLOW_V_2_10_PLUS = get_base_airflow_version_tuple() >= (2, 10, 0)
 AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)
+AIRFLOW_V_3_0 = get_base_airflow_version_tuple() == (3, 0, 0)

providers/microsoft/azure/tests/unit/microsoft/azure/log/test_wasb_task_handler.py

@@ -32,7 +32,7 @@
 
 from tests_common.test_utils.config import conf_vars
 from tests_common.test_utils.db import clear_db_dags, clear_db_runs
-from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_0, AIRFLOW_V_3_0_PLUS
 
 pytestmark = pytest.mark.db_test
 
@@ -122,10 +122,16 @@ def test_wasb_read(self, mock_hook_cls, ti):
             assert logs[0].sources == ["https://wasb-container.blob.core.windows.net/abc/hello.log"]
             assert logs[1].event == "::endgroup::"
             assert logs[2].event == "Log line"
-            assert metadata == {
-                "end_of_log": True,
-                "first_time_read": False,
-            }
+            if AIRFLOW_V_3_0:
+                assert metadata == {
+                    "end_of_log": True,
+                    "log_pos": 1,
+                }
+            else:
+                assert metadata == {
+                    "end_of_log": True,
+                    "first_time_read": False,
+                }
         else:
             assert logs[0][0][0] == "localhost"
             assert (

providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py

@@ -36,7 +36,10 @@
 from airflow.models import DagRun
 from airflow.providers.opensearch.log.os_json_formatter import OpensearchJSONFormatter
 from airflow.providers.opensearch.log.os_response import Hit, OpensearchResponse
-from airflow.providers.opensearch.version_compat import AIRFLOW_V_3_0_PLUS
+from airflow.providers.opensearch.version_compat import (
+    AIRFLOW_V_3_0,
+    AIRFLOW_V_3_0_PLUS,
+)
 from airflow.utils import timezone
 from airflow.utils.log.file_task_handler import FileTaskHandler
 from airflow.utils.log.logging_mixin import ExternalLoggingMixin, LoggingMixin
@@ -391,7 +394,7 @@ def _read(
                     "If your task started recently, please wait a moment and reload this page. "
                     "Otherwise, the logs for this task instance may have been removed."
                 )
-                if AIRFLOW_V_3_0_PLUS:
+                if AIRFLOW_V_3_0_PLUS and not AIRFLOW_V_3_0:  # over 3.0.0
                     from airflow.utils.log.file_task_handler import (
                         StructuredLogMessage,
                         get_compatible_output_log_stream,
@@ -423,7 +426,7 @@ def concat_logs(hits: list[Hit]):
             return "\n".join(self._format_msg(hits[i]) for i in range(log_range))
 
         if logs_by_host:
-            if AIRFLOW_V_3_0_PLUS:
+            if AIRFLOW_V_3_0_PLUS and not AIRFLOW_V_3_0:  # over 3.0.0
                 from airflow.utils.log.file_task_handler import (
                     StructuredLogMessage,
                     get_compatible_output_log_stream,

providers/opensearch/src/airflow/providers/opensearch/version_compat.py

@@ -34,3 +34,4 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
 
 AIRFLOW_V_2_10_PLUS = get_base_airflow_version_tuple() >= (2, 10, 0)
 AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)
+AIRFLOW_V_3_0 = get_base_airflow_version_tuple() == (3, 0, 0)