Fix 3.0 compat test for cloudwatch, s3

Author: jason810496

providers/amazon/tests/unit/amazon/aws/log/test_cloudwatch_task_handler.py

@@ -44,7 +44,7 @@
 from airflow.utils.timezone import datetime
 
 from tests_common.test_utils.config import conf_vars
-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
 
 
 def get_time_str(time_in_milliseconds):
@@ -259,13 +259,13 @@ def test_read(self, monkeypatch):
                 {"timestamp": current_time, "message": "Third"},
             ],
         )
-        if AIRFLOW_V_3_0_PLUS:
+        if AIRFLOW_V_3_0_PLUS and not AIRFLOW_V_3_0:  # 3.0+
             monkeypatch.setattr(
                 self.cloudwatch_task_handler,
                 "_read_from_logs_server",
                 lambda ti, worker_log_rel_path, log_metadata: ([], []),
             )
-        else:
+        else:  # 2.x and 3.0
             monkeypatch.setattr(
                 self.cloudwatch_task_handler,
                 "_read_from_logs_server",
@@ -299,7 +299,10 @@ def test_read(self, monkeypatch):
                     "timestamp": pendulum.datetime(2025, 3, 27, 21, 58, 1),
                 },
             ]
-            assert not metadata["first_time_read"]
+            if AIRFLOW_V_3_0:
+                assert metadata == {"end_of_log": False, "log_pos": 3}
+            else:
+                assert metadata == {"end_of_log": True, "first_time_read": False}
         else:
             events = "\n".join(
                 [

providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py

@@ -35,7 +35,7 @@
 from airflow.utils.timezone import datetime
 
 from tests_common.test_utils.config import conf_vars
-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
 
 
 @pytest.fixture(autouse=True)
@@ -256,7 +256,10 @@ def test_read(self):
             assert log[3].event == "Line 2"
             assert log[4].event == "Log line 3"
             assert log[5].event == "Line 4"
-            assert metadata == {"end_of_log": True, "first_time_read": False}
+            if AIRFLOW_V_3_0:
+                assert metadata == {"end_of_log": True, "log_pos": 4}
+            else:
+                assert metadata == {"end_of_log": True, "first_time_read": False}
         else:
             actual = log[0][0][-1]
             assert f"*** Found logs in s3:\n***   * {expected_s3_uri}\n" in actual
@@ -271,7 +274,10 @@ def test_read_when_s3_log_missing(self):
         if AIRFLOW_V_3_0_PLUS:
             log = list(log)
             assert len(log) == 2
-            assert metadata == {"end_of_log": True, "first_time_read": False}
+            if AIRFLOW_V_3_0:
+                assert metadata == {"end_of_log": True, "log_pos": 0}
+            else:
+                assert metadata == {"end_of_log": True, "first_time_read": False}
         else:
             assert len(log) == 1
             assert len(log) == len(metadata)