20260402 add ai edited test

tests/single_card_tests/ai_edited_test/config_and_utils/readme.md

@@ -0,0 +1,20 @@
+# Config and Utils Tests
+
+本目录包含 PaddleFleet 配置日志和工具模块的 AI 自动生成单元测试。
+
+This directory contains AI-generated unit tests for PaddleFleet config logger and utility modules.
+
+## Tested Files
+
+| Source File | Test File | Coverage Target |
+|-------------|-----------|----------------|
+| `config_logger.py` | `test_ai_config_logger.py` | get_config_logger_path, JSONEncoderWithMcoreTypes, log_config_to_disk |
+| `spec_utils.py` | `test_ai_spec_utils.py` | LayerSpec, import_layer, get_layer, build_layer |
+| `timers.py` | `test_ai_timers.py` | _Timer, RuntimeTimer, Timers |
+
+## Test Categories
+
+- JSON encoding of Paddle types (dtype, nn.Module, dict, list)
+- Disk logging with config paths and rank strings
+- Dynamic layer import and instantiation
+- Timer start/stop/reset/elapsed operations

tests/single_card_tests/ai_edited_test/config_and_utils/test_ai_config_logger.py

@@ -0,0 +1,153 @@
+# 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 json
+import os
+import tempfile
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import paddle
+from paddle import nn
+
+
+class TestConfigLogger(unittest.TestCase):
+    """Tests for config_logger module."""
+
+    def test_get_config_logger_path(self):
+        from paddlefleet.config_logger import get_config_logger_path
+
+        config = SimpleNamespace(config_logger_dir="/tmp/test_logger")
+        path = get_config_logger_path(config)
+        self.assertEqual(path, "/tmp/test_logger")
+
+    def test_get_config_logger_path_default(self):
+        from paddlefleet.config_logger import get_config_logger_path
+
+        config = SimpleNamespace()
+        path = get_config_logger_path(config)
+        self.assertEqual(path, "")
+
+    def test_has_config_logger_enabled(self):
+        from paddlefleet.config_logger import has_config_logger_enabled
+
+        config = SimpleNamespace(config_logger_dir="/tmp/test")
+        self.assertTrue(has_config_logger_enabled(config))
+
+    def test_has_config_logger_disabled(self):
+        from paddlefleet.config_logger import has_config_logger_enabled
+
+        config = SimpleNamespace()
+        self.assertFalse(has_config_logger_enabled(config))
+
+    def test_get_path_count_first(self):
+        from paddlefleet.config_logger import (
+            get_path_count,
+        )
+
+        count = get_path_count("/tmp/test_unique_3")
+        self.assertEqual(count, 0)
+
+    def test_get_path_count_increments(self):
+        from paddlefleet.config_logger import get_path_count
+
+        get_path_count("/tmp/test_unique_4")
+        count1 = get_path_count("/tmp/test_unique_4")
+        self.assertEqual(count1, 1)
+
+    def test_get_path_with_count(self):
+        from paddlefleet.config_logger import get_path_with_count
+
+        path = get_path_with_count("/tmp/test_unique_5")
+        self.assertTrue(path.endswith(".iter0"))
+
+    def test_json_encoder_paddle_dtype(self):
+        from paddlefleet.config_logger import JSONEncoderWithMcoreTypes
+
+        result = json.dumps(paddle.float32, cls=JSONEncoderWithMcoreTypes)
+        self.assertIn("float32", result)
+
+    def test_json_encoder_nn_module(self):
+        from paddlefleet.config_logger import JSONEncoderWithMcoreTypes
+
+        module = nn.Linear(10, 5)
+        result = json.dumps(module, cls=JSONEncoderWithMcoreTypes)
+        self.assertIsNotNone(result)
+
+    def test_json_encoder_dict(self):
+        from paddlefleet.config_logger import JSONEncoderWithMcoreTypes
+
+        data = {"a": 1, "b": paddle.float32}
+        result = json.dumps(data, cls=JSONEncoderWithMcoreTypes)
+        self.assertIn("float32", result)
+
+    def test_json_encoder_list(self):
+        from paddlefleet.config_logger import JSONEncoderWithMcoreTypes
+
+        data = [paddle.float16, paddle.float32]
+        result = json.dumps(data, cls=JSONEncoderWithMcoreTypes)
+        self.assertIsNotNone(result)
+
+    def test_json_encoder_string_fallback(self):
+        from paddlefleet.config_logger import JSONEncoderWithMcoreTypes
+
+        class CustomObj:
+            pass
+
+        result = json.dumps(CustomObj(), cls=JSONEncoderWithMcoreTypes)
+        self.assertIsNotNone(result)
+
+    @patch("paddlefleet.config_logger.parallel_state")
+    def test_log_config_to_disk(self, mock_ps):
+        from paddlefleet.config_logger import log_config_to_disk
+
+        mock_ps.get_all_ranks = MagicMock(return_value="0_0")
+        with tempfile.TemporaryDirectory() as tmpdir:
+            config = SimpleNamespace(config_logger_dir=tmpdir)
+            data = {"key": "value", "dtype": paddle.float32}
+            log_config_to_disk(config, data, prefix="test", rank_str="0_0")
+            files = os.listdir(tmpdir)
+            self.assertTrue(any(f.startswith("test") for f in files))
+
+    @patch("paddlefleet.config_logger.parallel_state")
+    def test_log_config_to_disk_with_self(self, mock_ps):
+        from paddlefleet.config_logger import log_config_to_disk
+
+        mock_ps.get_all_ranks = MagicMock(return_value="0_0")
+        with tempfile.TemporaryDirectory() as tmpdir:
+            config = SimpleNamespace(config_logger_dir=tmpdir)
+            m = nn.Linear(5, 3)
+            data = {"self": m, "extra": 42}
+            log_config_to_disk(config, data, prefix="layer", rank_str="0")
+            files = os.listdir(tmpdir)
+            self.assertTrue(len(files) > 0)
+
+    @patch("paddlefleet.config_logger.parallel_state")
+    def test_log_config_to_disk_ordereddict(self, mock_ps):
+        from collections import OrderedDict
+
+        from paddlefleet.config_logger import log_config_to_disk
+
+        mock_ps.get_all_ranks = MagicMock(return_value="0_0")
+        with tempfile.TemporaryDirectory() as tmpdir:
+            config = SimpleNamespace(config_logger_dir=tmpdir)
+            data = OrderedDict([("a", 1), ("b", 2)])
+            log_config_to_disk(config, data, prefix="od_test", rank_str="0")
+            files = os.listdir(tmpdir)
+            self.assertTrue(any(".pth" in f for f in files))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/single_card_tests/ai_edited_test/config_and_utils/test_ai_spec_utils.py

@@ -0,0 +1,120 @@
+# 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 unittest
+
+import paddle
+
+
+class TestLayerSpec(unittest.TestCase):
+    """Tests for LayerSpec dataclass."""
+
+    def test_repr_with_type(self):
+        from paddlefleet.spec_utils import LayerSpec
+
+        spec = LayerSpec(layer=paddle.nn.Linear)
+        r = repr(spec)
+        self.assertIn("Linear", r)
+
+    def test_default_kwargs(self):
+        from paddlefleet.spec_utils import LayerSpec
+
+        spec = LayerSpec(layer=paddle.nn.Linear)
+        self.assertEqual(spec.extra_kwargs, {})
+
+    def test_with_extra_kwargs(self):
+        from paddlefleet.spec_utils import LayerSpec
+
+        spec = LayerSpec(layer=paddle.nn.Linear, extra_kwargs={"bias": False})
+        self.assertEqual(spec.extra_kwargs, {"bias": False})
+
+
+class TestImportLayer(unittest.TestCase):
+    """Tests for import_layer function."""
+
+    def test_valid_import(self):
+        from paddlefleet.spec_utils import import_layer
+
+        layer = import_layer(("paddle.nn", "Linear"))
+        self.assertIs(layer, paddle.nn.Linear)
+
+    def test_invalid_import(self):
+        from paddlefleet.spec_utils import import_layer
+
+        layer = import_layer(("nonexistent_module_xyz", "FakeClass"))
+        self.assertIsNone(layer)
+
+
+class TestGetLayer(unittest.TestCase):
+    """Tests for get_layer function."""
+
+    def test_with_type(self):
+        from paddlefleet.spec_utils import get_layer
+
+        layer = get_layer(paddle.nn.Linear)
+        self.assertIs(layer, paddle.nn.Linear)
+
+    def test_with_function(self):
+        from paddlefleet.spec_utils import get_layer
+
+        def dummy_func():
+            pass
+
+        layer = get_layer(dummy_func)
+        self.assertIs(layer, dummy_func)
+
+    def test_with_spec_type(self):
+        from paddlefleet.spec_utils import LayerSpec, get_layer
+
+        spec = LayerSpec(layer=paddle.nn.Linear)
+        layer = get_layer(spec)
+        self.assertIs(layer, paddle.nn.Linear)
+
+    def test_with_spec_tuple(self):
+        from paddlefleet.spec_utils import LayerSpec, get_layer
+
+        spec = LayerSpec(layer=("paddle.nn", "Linear"))
+        layer = get_layer(spec)
+        self.assertIs(layer, paddle.nn.Linear)
+
+
+class TestBuildLayer(unittest.TestCase):
+    """Tests for build_layer function."""
+
+    def test_with_type(self):
+        from paddlefleet.spec_utils import build_layer
+
+        layer = build_layer(paddle.nn.Linear, 10, 5)
+        self.assertIsInstance(layer, paddle.nn.Linear)
+        self.assertEqual(layer.weight.shape[1], 5)
+
+    def test_with_spec_and_kwargs(self):
+        from paddlefleet.spec_utils import LayerSpec, build_layer
+
+        spec = LayerSpec(layer=paddle.nn.ReLU, extra_kwargs={})
+        layer = build_layer(spec)
+        self.assertIsInstance(layer, paddle.nn.ReLU)
+
+    def test_with_function(self):
+        from paddlefleet.spec_utils import build_layer
+
+        def my_func(x):
+            return x
+
+        result = build_layer(my_func)
+        self.assertIs(result, my_func)
+
+
+if __name__ == "__main__":
+    unittest.main()