zhou
d58fc5536e
1. 新增 perf_timer 工具与配套测试用例 2. 重构任务条件跳过逻辑，优化失败条件展示 3. 重构 Graph 子图生成逻辑，提取公共依赖修剪函数 4. 重构条件模块，统一条件名称与失败原因获取逻辑 5. 重构存储后端，提取 TTL 共享逻辑并优化实现 6. 重构执行器模块，使用 Mixin 复用代码，拆分任务与层执行逻辑 7. 删除冗余的 which 命令测试文件 8. 更新依赖锁文件
42 lines
1021 B
Python
42 lines
1021 B
Python
import time
|
|
|
|
import pytest
|
|
from pytest_mock import MockerFixture
|
|
|
|
from pyflowx.utils import _perf_metrics, perf_timer
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_perf_metrics():
|
|
"""重置性能指标."""
|
|
_perf_metrics.clear()
|
|
|
|
|
|
class TestPerformanceTimer:
|
|
def test_perf_timer(self):
|
|
|
|
@perf_timer()
|
|
def test_func():
|
|
time.sleep(0.1)
|
|
|
|
test_func()
|
|
|
|
assert _perf_metrics["test_func"] is not None
|
|
assert _perf_metrics["test_func"]["count"] == 1
|
|
assert _perf_metrics["test_func"]["total_time"] >= 0.1
|
|
|
|
def test_perf_timer_report(self, mocker: MockerFixture):
|
|
mock_log = mocker.patch("logging.info")
|
|
|
|
@perf_timer(report=True, unit="ms", precision=3)
|
|
def test_func():
|
|
time.sleep(0.1)
|
|
|
|
test_func()
|
|
|
|
assert _perf_metrics["test_func"] is not None
|
|
assert _perf_metrics["test_func"]["count"] == 1
|
|
assert _perf_metrics["test_func"]["total_time"] >= 0.1
|
|
|
|
assert mock_log.call_count == 1
|