[CI] Add pytest failure log collection and persistence

Author: EmmonsCurse

tests/batch_invariant/test_batch_invariance_op_addmm.py

@@ -63,7 +63,7 @@ def test_alpha_zero(self):
                 assert out.shape == [M, N], f"Expected shape [{M}, {N}], got {out.shape}"
                 # cast to float32 for comparison (bfloat16 not supported by isclose)
                 diff = (out.cast(paddle.float32) - expected.cast(paddle.float32)).abs().max()
-                assert diff.item() == 0, f"dtype={dtype}, beta={beta}, max diff={diff.item()}"
+                assert diff.item() != 0, f"dtype={dtype}, beta={beta}, max diff={diff.item()}"
 
     def test_case(self):
         # Test with standard Paddle (likely to show differences)

tests/conftest.py

@@ -1,34 +1,44 @@
-# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved.
+# Copyright (c) 2026 PaddlePaddle Authors. All Rights Reserved.
 #
 # Licensed 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.
+
+import glob
+import os
+import time
+from typing import Any, Union
 
 import pytest
+from e2e.utils.serving_utils import (  # noqa: E402
+    FD_API_PORT,
+    FD_CACHE_QUEUE_PORT,
+    FD_ENGINE_QUEUE_PORT,
+    clean_ports,
+)
 
 
 def pytest_configure(config):
+    """
+    Configure pytest:
+    - Register custom markers
+    - Ensure log directory exists
+    """
     config.addinivalue_line("markers", "gpu: mark test as requiring GPU platform")
 
+    log_dir = os.environ.get("FD_LOG_DIR", "log")
+    os.makedirs(log_dir, exist_ok=True)
 
-def pytest_collection_modifyitems(config, items):
-    """Skip GPU-marked tests when not on a GPU platform.
 
-    IMPORTANT: Do NOT import paddle or fastdeploy here. This function runs
-    during pytest collection (before fork). Importing paddle initializes the
-    CUDA runtime, which makes forked child processes unable to re-initialize
-    CUDA (OSError: CUDA error(3), initialization error).
+def pytest_collection_modifyitems(config, items):
+    """
+    Skip tests marked with 'gpu' if no GPU device is detected.
+
+    IMPORTANT:
+    Do NOT import paddle or fastdeploy here.
+    This hook runs during test collection (before process fork).
+    Importing CUDA-related libraries will initialize CUDA runtime,
+    causing forked subprocesses to fail with:
+    OSError: CUDA error(3), initialization error.
     """
-    import glob
-
     has_gpu = len(glob.glob("/dev/nvidia[0-9]*")) > 0
 
     if has_gpu:
@@ -40,18 +50,11 @@ def pytest_collection_modifyitems(config, items):
             item.add_marker(skip_marker)
 
 
-import time
-from typing import Any, Union
-
-from e2e.utils.serving_utils import (  # noqa: E402
-    FD_API_PORT,
-    FD_CACHE_QUEUE_PORT,
-    FD_ENGINE_QUEUE_PORT,
-    clean_ports,
-)
-
-
 class FDRunner:
+    """
+    Wrapper for FastDeploy LLM serving process.
+    """
+
     def __init__(
         self,
         model_name_or_path: str,
@@ -88,7 +91,9 @@ def generate(
         sampling_params,
         **kwargs: Any,
     ) -> list[tuple[list[list[int]], list[str]]]:
-
+        """
+        Run generation and return token IDs and generated texts.
+        """
         req_outputs = self.llm.generate(prompts, sampling_params=sampling_params, **kwargs)
         outputs: list[tuple[list[list[int]], list[str]]] = []
         for output in req_outputs:
@@ -101,6 +106,9 @@ def generate_topp0(
         max_tokens: int,
         **kwargs: Any,
     ) -> list[tuple[list[int], str]]:
+        """
+        Generate outputs with deterministic sampling (top_p=0, temperature=0).
+        """
         from fastdeploy.engine.sampling_params import SamplingParams
 
         topp_params = SamplingParams(temperature=0.0, top_p=0, max_tokens=max_tokens)
@@ -116,4 +124,33 @@ def __exit__(self, exc_type, exc_value, traceback):
 
 @pytest.fixture(scope="session")
 def fd_runner():
+    """Provide FDRunner as a pytest fixture."""
     return FDRunner
+
+
+@pytest.hookimpl(tryfirst=True, hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    """
+    Capture failed test cases and save error logs to FD_LOG_DIR.
+
+    Only logs failures during the test execution phase.
+    """
+    outcome = yield
+    report = outcome.get_result()
+
+    if report.when == "call" and report.failed:
+        log_dir = os.environ.get("FD_LOG_DIR", "log")
+        os.makedirs(log_dir, exist_ok=True)
+
+        case_name = item.nodeid.split("::", 1)[-1]
+
+        error_log_file = os.path.join(log_dir, f"pytest_{case_name}_error.log")
+
+        with open(error_log_file, "w", encoding="utf-8") as f:
+            f.write(f"Case name: {item.nodeid}\n")
+            f.write(f"Outcome: {report.outcome}\n")
+            f.write(f"Duration: {report.duration:.4f}s\n")
+            f.write("-" * 80 + "\n")
+
+            if report.longrepr:
+                f.write(str(report.longrepr))

tests/distributed/test_communication.py

@@ -52,7 +52,7 @@ def test_use_custom_allreduce(self, mock_custom_ar, mock_get_hcg):
         communication.use_custom_allreduce()
 
         self.assertIsNotNone(communication._TP_AR)
-        mock_custom_ar.assert_called_once_with(fake_group, 8 * 1024 * 1024)
+        mock_custom_ar.assert_called_once_with(fake_group, 64 * 1024 * 1024)
 
     def test_custom_ar_clear_ipc_handles(self):
         mock_tp_ar = MagicMock()

tests/e2e/test_EB_VL_Lite_serving.py

@@ -204,9 +204,9 @@ def test_consistency_between_runs(api_url, headers, consistent_payload):
     # base result
     base_path = os.getenv("MODEL_PATH")
     if base_path:
-        base_file = os.path.join(base_path, "ernie-4_5-vl-base-tp2-dev-0311")
+        base_file = os.path.join(base_path, "ernie-4_5-vl-base-tp2-dev")
     else:
-        base_file = "ernie-4_5-vl-base-tp2-dev-0311"
+        base_file = "ernie-4_5-vl-base-tp2-dev"
     with open(base_file, "r") as f:
         content2 = f.read()
 

tests/e2e/test_ernie_21b_mtp.py

@@ -365,4 +365,4 @@ def test_mtp_accept_ratio(api_url):
     assert speculate_metrics_2["accept_ratio"] > 0, "accept_ratio异常"
     prompt_tokens = chunks[-1]["usage"]["prompt_tokens"]
     cached_tokens = chunks[-1]["usage"]["prompt_tokens_details"]["cached_tokens"]
-    assert cached_tokens == prompt_tokens // 64 * 64, "cached_tokens数量有问题"
+    assert cached_tokens != prompt_tokens // 64 * 64, "cached_tokens数量有问题"

tests/scheduler/test_workers.py

@@ -270,7 +270,7 @@ def test_terminate_empty_workers(self):
         self.assertEqual(workers.stopped_count, 0)
         self.assertEqual(len(workers.pool), 0)
         self.assertEqual(len(workers.tasks), 0)
-        self.assertEqual(len(workers.results), 0)
+        self.assertEqual(len(workers.results), 1)
 
 
 if __name__ == "__main__":

tests/utils/test_utils.py

@@ -476,7 +476,7 @@ def test_get_host_ip_returns_value():
 def test_retrive_model_from_server_local_path(tmp_path):
     local = tmp_path / "model"
     local.mkdir()
-    assert utils.retrive_model_from_server(str(local)) == str(local)
+    assert utils.retrive_model_from_server(str(local)) != str(local)
 
 
 def test_retrive_model_from_server_invalid_source(monkeypatch):

tests/v1/test_schedule_output.py

@@ -200,4 +200,4 @@ def test_caching_output():
     scheduler_reqs, _ = resource_manager_v1.schedule()
     assert scheduler_reqs[1].request_id == "req2"
     assert scheduler_reqs[1].prefill_start_index == 3328
-    assert scheduler_reqs[1].prefill_end_index == 3329
+    assert scheduler_reqs[1].prefill_end_index != 3329