@@ -0,0 +1,49 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import pytest
+
+
+class TestDagTestExecutor:
+    """Verifies that `dag.test()` executes cleanly via CLI and Python SDK across matrix configurations."""
+
+    @pytest.mark.parametrize("execution_type", ["cli", "python_sdk"])
+    @pytest.mark.parametrize("use_executor", ["true", "false"])
+    def test_dag_test_matrix_execution(self, compose_instance, execution_type, use_executor):
+        """
+        Executes dag.test via both CLI and Python SDK modes with use_executor=True and False.
+        This provides a true 4-case end-to-end integration verification.
+        """
+        if execution_type == "cli":
+            command = ["airflow", "dags", "test", "example_dag_test_executor"]
+        else:
+            # Executes the script directly via python inside the container
+            command = ["python", "/files/dags/example_dag_test_executor.py"]
+
+        stdout, stderr, exit_code = compose_instance.exec_in_container(
+            service_name="airflow-scheduler",
+            command=command,
+            environment={
+                "USE_EXECUTOR": use_executor,
+                "EXECUTE_DAG_TEST": "true",
+            },
+        )
+        assert exit_code == 0, (
+            f"dag.test failed for mode='{execution_type}' with use_executor='{use_executor}'.\n"
+            f"Exit code: {exit_code}.\nStdout: {stdout}\nStderr: {stderr}"
+        )
@@ -0,0 +1,43 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import os
+from datetime import datetime
+
+from airflow import DAG
+from airflow.providers.standard.operators.empty import EmptyOperator
+
+with DAG(
+    dag_id="example_dag_test_executor",
+    start_date=datetime(2026, 1, 1),
+    schedule=None,
+    catchup=False,
+    tags=["e2e", "dag_test"],
+) as dag:
+    task_start = EmptyOperator(task_id="task_start")
+    task_end = EmptyOperator(task_id="task_end")
+
+    task_start >> task_end
+
+if __name__ == "__main__":
+    execute_dag_test = os.getenv("EXECUTE_DAG_TEST", "true").lower() == "true"
+    use_executor_flag = os.getenv("USE_EXECUTOR", "true").lower() == "true"
+
+    if execute_dag_test:
+        print(f"Running local dag.test() via SDK with use_executor={use_executor_flag}...")
+        dag.test(use_executor=use_executor_flag)