diff --git a/Makefile b/Makefile index 06f0ef7b6..5fe6c1b50 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: init sync format proto unit bench +.PHONY: init sync format proto unit bench clean init: uv sync --all-extras @@ -18,4 +18,7 @@ unit: uv run pytest tests/unit bench: - uv run pytest tests/benchmark \ No newline at end of file + uv run pytest tests/benchmark + +clean: + @bash scripts/clean_pycache.sh . \ No newline at end of file diff --git a/scripts/clean_pycache.sh b/scripts/clean_pycache.sh new file mode 100755 index 000000000..2f40d1d21 --- /dev/null +++ b/scripts/clean_pycache.sh @@ -0,0 +1,39 @@ +#!/bin/bash +# Description: 递归清除指定目录下的 __pycache__ 和 .pyc 文件 +# 如果删除 __pycache__ 后导致其父目录为空,则连同由于删除产生的空父目录一起递归向上删除 + +# 目标目录,默认当前目录 +TARGET_DIR="${1:-.}" + +echo "开始清理 $TARGET_DIR 目录下的 Python 缓存文件..." + +# 1. 寻找所有的 __pycache__ 文件夹,并循环处理 +# 注意:find 结果通过 while 读取,处理带有空格等特殊符号的路径 +find "$TARGET_DIR" -type d -name "__pycache__" | while read -r cache_dir; do + # 获取 __pycache__ 所在的父目录 + parent_dir="$(dirname "$cache_dir")" + + # 删除该 __pycache__ 目录及里面的缓存文件 + rm -rf "$cache_dir" + + echo "已清理: $cache_dir" + + # 尝试递归往上删除空的父目录 + # rmdir 的特性是:只有在此目录没有任何文件/文件夹时才会删除并返回成功 + # 因此我们循环往上 rmdir,如果哪个父层级除了 __pycache__ 还有其它文件,自然会失败并跳出循环 + while [ "$parent_dir" != "." ] && [ "$parent_dir" != "/" ] && [ "$parent_dir" != "$TARGET_DIR" ]; do + if rmdir "$parent_dir" 2>/dev/null; then + echo "已删除原本仅含 __pycache__ 的空父目录: $parent_dir" + # 指向更上层父目录继续检查 + parent_dir="$(dirname "$parent_dir")" + else + # 遇到了非空目录(包含了其它文件),终止向上删除 + break + fi + done +done + +# 2. 清理散落在其它地方的独立 .pyc 后缀缓存文件 +find "$TARGET_DIR" -type f -name "*.pyc" -delete 2>/dev/null + +echo "清理完成!" diff --git a/swanlab/sdk/internal/core.py b/swanlab/sdk/internal/core.py index b0bd82a08..174276438 100644 --- a/swanlab/sdk/internal/core.py +++ b/swanlab/sdk/internal/core.py @@ -6,6 +6,7 @@ """ from abc import ABC, abstractmethod +from enum import Enum from typing import List from swanlab.proto.swanlab.record.v1.record_pb2 import Record @@ -14,6 +15,11 @@ __all__ = ["CoreProtocol"] +class CoreEnum(str, Enum): + CORE_PYTHON = "CorePython" + CORE = "Core" + + class CoreProtocol(ABC): """ SwanLab Core 的抽象基类。 @@ -47,3 +53,13 @@ def handle_records(self, records: List[Record]) -> None: def shutdown(self) -> None: """刷盘、触发收尾回调、释放所有资源。调用后此实例不应再被使用。""" ... + + +def create_core(ctx: RunContext, core_enum: CoreEnum = CoreEnum.CORE_PYTHON) -> CoreProtocol: + if core_enum == CoreEnum.CORE_PYTHON: + from swanlab.sdk.internal.core_python import CorePython + + return CorePython(ctx) + else: + # TODO: Core 微服务无感接入 + raise NotImplementedError diff --git a/swanlab/sdk/internal/core_python/__init__.py b/swanlab/sdk/internal/core_python/__init__.py index 86056e4e6..eaf032947 100644 --- a/swanlab/sdk/internal/core_python/__init__.py +++ b/swanlab/sdk/internal/core_python/__init__.py @@ -2,6 +2,7 @@ @author: cunyue @file: __init__.py @time: 2026/3/13 + @description: SwanLab Core Python 版本,封装SwanLab云端版核心业务,包括: 1. 提供http客户端,用于与SwanLab云端API进行交互。 2. 提供rpc封装函数,以rpc方式调用SwanLab云端API。 @@ -13,12 +14,13 @@ BackgroundConsumer 等调用方无需修改。 """ -from typing import List +from typing import List, Optional from swanlab.proto.swanlab.record.v1.record_pb2 import Record from swanlab.sdk.internal.context import RunContext from swanlab.sdk.internal.core import CoreProtocol from swanlab.sdk.internal.core_python.store import DataStoreWriter +from swanlab.sdk.internal.core_python.uploader import Uploader from swanlab.sdk.internal.pkg import console from swanlab.sdk.utils.helper.env import DEBUG @@ -33,26 +35,34 @@ class CorePython(CoreProtocol): def __init__(self, ctx: RunContext): super().__init__(ctx) - self._store: DataStoreWriter | None = None + self._store: Optional[DataStoreWriter] = None + self._uploader: Optional[Uploader] = None def startup(self, cloud: bool, persistence: bool) -> None: - if self._store is not None: + if self._store is not None or self._uploader is not None: raise RuntimeError("CorePython has already been started.") if persistence: self._store = DataStoreWriter() self._store.open(str(self._ctx.run_file)) + if cloud: + self._uploader = Uploader() def handle_records(self, records: List[Record]) -> None: - if self._store is None: + if self._store is None and self._uploader is None: console.warning("CorePython is not started, skipping record handling.") return - for record in records: - self._store.write(record.SerializeToString()) - if DEBUG: - console.debug("Write record:", record.WhichOneof("record_type")) + if self._store is not None: + for record in records: + self._store.write(record.SerializeToString()) + if DEBUG: + console.debug("Write record:", record.WhichOneof("record_type")) + if self._uploader is not None: + self._uploader.put(records) def shutdown(self) -> None: - if self._store is None: - return - self._store.close() - self._store = None + if self._uploader is not None: + self._uploader.finish() + self._uploader = None + if self._store is not None: + self._store.close() + self._store = None diff --git a/swanlab/sdk/internal/core_python/uploader/__init__.py b/swanlab/sdk/internal/core_python/uploader/__init__.py new file mode 100644 index 000000000..b1a84d063 --- /dev/null +++ b/swanlab/sdk/internal/core_python/uploader/__init__.py @@ -0,0 +1,24 @@ +""" +@author: caddiesnew +@file: __init__.py +@time: 2026/3/19 +@description: 指标上传线程,基于 protobuf Record +""" + +from .collector import Collector +from .sender import ( + HttpRecordTransport, + create_record_transport, + trace_records, + upload_records, +) +from .thread import Uploader + +__all__ = [ + "Collector", + "HttpRecordTransport", + "Uploader", + "create_record_transport", + "trace_records", + "upload_records", +] diff --git a/swanlab/sdk/internal/core_python/uploader/collector.py b/swanlab/sdk/internal/core_python/uploader/collector.py new file mode 100644 index 000000000..ac9d54fcb --- /dev/null +++ b/swanlab/sdk/internal/core_python/uploader/collector.py @@ -0,0 +1,84 @@ +""" +@author: caddiesnew +@file: collector.py +@time: 2026/3/21 +@description: 上传线程采集器 +""" + +import threading +import time +from typing import Callable, List, Optional + +from swanlab.proto.swanlab.record.v1.record_pb2 import Record +from swanlab.sdk.internal.core_python.uploader.sender import upload_records +from swanlab.sdk.internal.pkg import console + + +class Collector: + """ + 日志采集器,负责从队列中收集 Records 并批量上传。 + """ + + def __init__( + self, + upload_interval: float = 1.0, + upload_callback: Optional[Callable] = None, + ): + self.container: List[Record] = [] + self._lock = threading.Lock() + self._upload_interval = upload_interval + self._upload_callback = upload_callback + self._last_upload_at = time.time() + + def upload(self, pending: List[Record]) -> None: + """ + 核心上传逻辑:直接批量上送 protobuf Record。 + 由于网络请求耗时,应在无锁状态下调用。 + """ + if not pending: + return + upload_records(pending, upload_callback=self._upload_callback) + + def submit(self, records: List[Record]) -> None: + """提交记录,由定时线程以 upload_interval 为周期进行调用""" + with self._lock: + self.container.extend(records) + if not self.container: + return + + pending = self.container + if self._upload_interval <= 0: + self.container = [] + else: + now = time.time() + if now - self._last_upload_at <= self._upload_interval: + return + + self._last_upload_at = now + self.container = [] + + try: + self.upload(pending) + except Exception as exc: + console.error(f"upload error: {exc}") + with self._lock: + self.container = pending + self.container + + def flush(self, records: List[Record]) -> None: + """刷新所有累积的 records 进行上传""" + with self._lock: + self.container.extend(records) + pending = self.container + self.container = [] + + try: + self.upload(pending) + except Exception as exc: + console.error(f"upload error: {exc}") + with self._lock: + self.container = pending + self.container + + +__all__ = [ + "Collector", +] diff --git a/swanlab/sdk/internal/core_python/uploader/helper.py b/swanlab/sdk/internal/core_python/uploader/helper.py new file mode 100644 index 000000000..3244ff480 --- /dev/null +++ b/swanlab/sdk/internal/core_python/uploader/helper.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +r""" +@author: caddiesnew +@file: helper.py +@time: 2026/3/30 15:33 +@description: uploader 辅助函数 +""" + +from collections import OrderedDict +from typing import Dict, Iterator, List, Sequence, Tuple + +from swanlab.proto.swanlab.record.v1.record_pb2 import Record + + +def generate_chunks(records: Sequence[Record], per_request_len: int) -> Iterator[Tuple[Sequence[Record], int]]: + """ + 按固定大小对 Record 序列做切片。 + yield: (分片数据, 分片长度) + """ + if per_request_len == -1: + yield records, len(records) + return + if per_request_len <= 0: + raise ValueError(f"per_request_len must be -1 or > 0, got {per_request_len}") + + total = len(records) + if total <= per_request_len: + yield records, total + return + + for index in range(0, total, per_request_len): + chunk = records[index : index + per_request_len] + yield chunk, len(chunk) + + +def group_records_by_type(records: Sequence[Record]) -> Dict[str, List[Record]]: + """按照 record 的类型进行分组""" + grouped: Dict[str, List[Record]] = OrderedDict() + for record in records: + if not isinstance(record, Record): + raise TypeError(f"group_records_by_type only accepts Record instances, got {type(record).__name__}") + record_type = record.WhichOneof("record_type") + if record_type is None: + raise ValueError("Record has no active record_type and cannot be uploaded.") + grouped.setdefault(record_type, []).append(record) + return grouped diff --git a/swanlab/sdk/internal/core_python/uploader/sender.py b/swanlab/sdk/internal/core_python/uploader/sender.py new file mode 100644 index 000000000..df91c1cc3 --- /dev/null +++ b/swanlab/sdk/internal/core_python/uploader/sender.py @@ -0,0 +1,78 @@ +""" +@author: caddiesnew +@file: sender.py +@time: 2026/3/19 +@description: Record 上传入口与 Core sidecar transport +""" + +from typing import Callable, Optional, Sequence + +from swanlab.proto.swanlab.record.v1.record_pb2 import Record +from swanlab.sdk.internal.pkg import console + +from .helper import generate_chunks, group_records_by_type + + +class HttpRecordTransport: + """PLACEHOLDER: 基于现有 HTTP client 的上传 transport 。""" + + def upload_record_group(self, record_type: str, records: Sequence[Record]) -> None: + if len(records) == 0: + return + console.debug(f"HTTP upload skeleton is ready for record_type={record_type!r}, but request mapping is pending.") + + def close(self) -> None: + pass + + +def create_record_transport() -> HttpRecordTransport: + """创建 Record transport。""" + return HttpRecordTransport() + + +def trace_records( + records: Optional[Sequence[Record]] = None, + per_request_len: int = 10_000, + upload_callback: Optional[Callable[[int], None]] = None, + transport: Optional[HttpRecordTransport] = None, +) -> None: + """ + 分片上传 protobuf Record。 + + 线程层负责缓冲、聚合与重试;此层只做 Record 分片、按 record_type 分组和 transport 调用。 + """ + if records is None or len(records) == 0: + return + + owns_transport = transport is None + active_transport = transport or create_record_transport() + + try: + for chunk, chunk_len in generate_chunks(records, per_request_len): + for record_type, grouped_records in group_records_by_type(chunk).items(): + active_transport.upload_record_group(record_type, grouped_records) + if upload_callback: + upload_callback(chunk_len) + finally: + if owns_transport: + active_transport.close() + + +def upload_records( + records: Sequence[Record], + upload_callback: Optional[Callable[[int], None]] = None, + per_request_len: int = 10_000, +) -> None: + """上传一组 protobuf Record,不在此层做任何 JSON 化。""" + if len(records) == 0: + return console.debug("No records to upload.") + trace_records(records, per_request_len=per_request_len, upload_callback=upload_callback) + return None + + +__all__ = [ + "HttpRecordTransport", + "create_record_transport", + "trace_records", + "upload_records", +] diff --git a/swanlab/sdk/internal/core_python/uploader/thread.py b/swanlab/sdk/internal/core_python/uploader/thread.py new file mode 100644 index 000000000..a01a582ba --- /dev/null +++ b/swanlab/sdk/internal/core_python/uploader/thread.py @@ -0,0 +1,85 @@ +""" +@author: caddiesnew +@file: uploader.py +@time: 2026/3/21 +@description: 上传线程池 +""" + +from queue import Empty, Queue +from typing import Callable, List, Optional + +from swanlab.proto.swanlab.record.v1.record_pb2 import Record +from swanlab.sdk.internal.core_python.uploader.collector import Collector +from swanlab.sdk.internal.pkg.timer import Timer + + +class Uploader: + """ + 管理上传工作线程和通信管道,定时消费 BackgroundConsumer 生产的 records + """ + + UPLOAD_INTERVAL = 1.0 + UPLOAD_THREAD_NAME = "SwanLab·CorePython.Uploader" + + def __init__( + self, + upload_interval: Optional[float] = None, + upload_callback: Optional[Callable] = None, + auto_start: bool = True, + ): + resolved_upload_interval = self.UPLOAD_INTERVAL if upload_interval is None else upload_interval + self._record_queue: Queue = Queue() + self._collector = Collector( + upload_interval=resolved_upload_interval, + upload_callback=upload_callback, + ) + self._timer = Timer( + lambda: self._collector.submit(self._drain_records()), + interval=resolved_upload_interval, + immediate=True, + name=self.UPLOAD_THREAD_NAME, + ) + self._started = False + self._finished = False + + if auto_start: + self.start() + + def start(self) -> None: + """启动上传线程。""" + if self._started or self._finished: + return + self._timer.start() + self._started = True + + def put(self, records: List[Record]) -> None: + """ + 主线程调用:将一批 Records 投递到队列 + """ + if self._finished: + raise RuntimeError("Uploader has already been finished.") + + self._record_queue.put(records) + + def finish(self) -> None: + """停止线程池,执行最终 flush。""" + if self._finished: + return + + self._finished = True + self._timer.cancel() + self._timer.join(timeout=10) + self._collector.flush(self._drain_records()) + + def _drain_records(self) -> List[Record]: + records: List[Record] = [] + while True: + try: + records.extend(self._record_queue.get_nowait()) + except Empty: + return records + + +__all__ = [ + "Uploader", +] diff --git a/swanlab/sdk/internal/pkg/timer/__init__.py b/swanlab/sdk/internal/pkg/timer/__init__.py new file mode 100644 index 000000000..92afc9fd4 --- /dev/null +++ b/swanlab/sdk/internal/pkg/timer/__init__.py @@ -0,0 +1,150 @@ +""" +@author: caddiesnew +@file: __init__.py +@time: 2026-03-18 00:03:03 +@description: SwanLab SDK 轻量级定时任务模块 + +设计约束: +1. 使用守护线程执行周期任务,避免因忘记清理导致主进程无法退出 +2. cancel() 为协作式停止,只会阻止下一轮调度,不会强杀正在执行的任务 +3. 支持固定间隔和动态间隔策略;动态策略接收“已完成执行次数”作为输入 +4. start()/run() 可重复调用,但运行中再次调用只会告警,不会启动重复线程 +""" + +import threading +from typing import Callable, Optional, Union + +from swanlab.sdk.internal.pkg import console + +__all__ = ["Timer"] + +Interval = Union[int, float] +IntervalStrategy = Callable[[int], Union[int, float]] +IntervalValue = Union[Interval, IntervalStrategy] + + +class Timer: + """ + 轻量级周期任务调度器。 + + :param task: 需要周期执行的无参函数 + :param interval: 固定间隔(秒)或动态间隔策略。动态策略入参为当前已完成执行次数。 + :param immediate: 是否在启动后先立即执行一次任务,再进入下一轮等待。 + :param name: 后台线程名称,便于调试定位。 + """ + + def __init__( + self, + task: Callable[[], None], + *, + interval: IntervalValue, + immediate: bool = False, + name: str = "SwanLab·Timer", + ) -> None: + self._task = task + self._interval = interval + self._immediate = immediate + self._name = name + + self._lock = threading.Lock() + self._stop_event = threading.Event() + self._thread: Optional[threading.Thread] = None + self._count = 0 + + if not callable(interval): + self._normalize_interval(interval) + + def start(self) -> "Timer": + """ + 启动定时器。若当前已在运行,则只告警并返回自身。 + """ + with self._lock: + if self._thread is not None and self._thread.is_alive(): + console.warning("Timer already running") + return self + + self._stop_event.clear() + self._thread = threading.Thread(target=self._loop, name=self._name, daemon=True) + self._thread.start() + + return self + + def run(self) -> "Timer": + """ + 兼容旧接口,等价于 start()。 + """ + return self.start() + + def cancel(self) -> None: + """ + 发出停止信号。 + 当前正在执行的任务不会被中断,但后续轮次不会再被调度。 + """ + self._stop_event.set() + + def join(self, timeout: Optional[float] = None) -> None: + """ + 等待后台线程退出,通常配合 cancel() 使用以保证任务完整收尾。 + """ + with self._lock: + thread = self._thread + + if thread is not None and thread.is_alive(): + thread.join(timeout) + + @property + def execution_count(self) -> int: + """ + 已完成执行次数。 + """ + return self._count + + @property + def is_running(self) -> bool: + """ + 当前后台线程是否仍在运行。 + """ + with self._lock: + return self._thread is not None and self._thread.is_alive() + + def _loop(self) -> None: + if self._immediate and not self._stop_event.is_set(): + self._execute_once() + + while not self._stop_event.is_set(): + try: + sleep_time = self._resolve_interval() + except Exception as e: + console.error(f"Timer interval strategy error: {e}") + self._stop_event.set() + return + + if self._stop_event.wait(sleep_time): + return + + self._execute_once() + + def _execute_once(self) -> None: + try: + self._task() + except Exception as e: + console.error(f"Error executing task: {e}") + finally: + self._count += 1 + + def _resolve_interval(self) -> float: + interval = self._interval + if callable(interval): + interval = interval(self._count) + return self._normalize_interval(interval) + + @staticmethod + def _normalize_interval(interval: Union[int, float]) -> float: + if isinstance(interval, bool): + raise TypeError("Timer interval must be a positive number, not bool.") + + value = float(interval) + if value <= 0: + raise ValueError(f"Timer interval must be greater than 0, got {interval!r}.") + + return value diff --git a/swanlab/sdk/internal/run/__init__.py b/swanlab/sdk/internal/run/__init__.py index 6e024762b..f95c128d3 100644 --- a/swanlab/sdk/internal/run/__init__.py +++ b/swanlab/sdk/internal/run/__init__.py @@ -23,7 +23,7 @@ from swanlab.sdk.internal.bus import RunEmitter from swanlab.sdk.internal.bus.events import MetricLogEvent, RunFinishEvent, RunStartEvent, ScalarDefineEvent from swanlab.sdk.internal.context import RunContext -from swanlab.sdk.internal.core_python import CorePython +from swanlab.sdk.internal.core import create_core from swanlab.sdk.internal.pkg import console, log from swanlab.sdk.internal.run.config import ( SwanLabConfig as _SwanLabConfigClass, @@ -126,7 +126,7 @@ def __init__(self, ctx: RunContext): self._emitter = RunEmitter(maxsize=100_000) # Core:Record 落盘与后端交互的统一入口 - self._core = CorePython(ctx) + self._core = create_core(ctx) # 记录构建器:负责将事件转换为 Record self._builder = RecordBuilder(ctx) diff --git a/tests/unit/sdk/cmd/test_cmd_helper.py b/tests/unit/sdk/cmd/test_cmd_helper.py index 56f406a28..15e48a4e7 100644 --- a/tests/unit/sdk/cmd/test_cmd_helper.py +++ b/tests/unit/sdk/cmd/test_cmd_helper.py @@ -1,6 +1,6 @@ """ @author: cunyue -@file: test_helper.py +@file: test_cmd_helper.py @time: 2026/3/14 @description: 测试 swanlab.sdk.cmd.helper 中的装饰器 """ diff --git a/tests/unit/sdk/internal/core_python/test_core_python.py b/tests/unit/sdk/internal/core_python/test_core_python.py new file mode 100644 index 000000000..1834f2f64 --- /dev/null +++ b/tests/unit/sdk/internal/core_python/test_core_python.py @@ -0,0 +1,48 @@ +from unittest.mock import MagicMock, patch + +from google.protobuf.timestamp_pb2 import Timestamp + +from swanlab.proto.swanlab.metric.column.v1.column_pb2 import ColumnType +from swanlab.proto.swanlab.metric.data.v1.data_pb2 import DataRecord +from swanlab.proto.swanlab.metric.data.v1.scalar.scalar_pb2 import ScalarValue +from swanlab.proto.swanlab.record.v1.record_pb2 import Record +from swanlab.sdk.internal.context import RunConfig, RunContext +from swanlab.sdk.internal.core_python import CorePython +from swanlab.sdk.internal.settings import Settings + + +def make_scalar_record() -> Record: + timestamp = Timestamp() + timestamp.GetCurrentTime() + return Record( + metric=DataRecord( + key="train/loss", + step=1, + timestamp=timestamp, + type=ColumnType.COLUMN_TYPE_FLOAT, + scalar=ScalarValue(number=0.125), + ) + ) + + +def make_ctx(tmp_path) -> RunContext: + settings = Settings.model_validate({"mode": "cloud", "run": {"id": "test-run-id"}}) + return RunContext(RunConfig(run_dir=tmp_path, settings=settings)) + + +def test_core_python_cloud_mode_forwards_records_to_uploader(tmp_path): + ctx = make_ctx(tmp_path) + + with patch("swanlab.sdk.internal.core_python.Uploader") as mock_pool_cls: + uploader = MagicMock() + mock_pool_cls.return_value = uploader + + core = CorePython(ctx) + core.startup(cloud=True, persistence=False) + + record = make_scalar_record() + core.handle_records([record]) + uploader.put.assert_called_once_with([record]) + + core.shutdown() + uploader.finish.assert_called_once_with() diff --git a/tests/unit/sdk/internal/core_python/uploader/conftest.py b/tests/unit/sdk/internal/core_python/uploader/conftest.py new file mode 100644 index 000000000..9c6fe2396 --- /dev/null +++ b/tests/unit/sdk/internal/core_python/uploader/conftest.py @@ -0,0 +1,40 @@ +import pytest +from google.protobuf.timestamp_pb2 import Timestamp + +from swanlab.proto.swanlab.config.v1.config_pb2 import ConfigRecord, UpdateType +from swanlab.proto.swanlab.metric.column.v1.column_pb2 import ColumnType +from swanlab.proto.swanlab.metric.data.v1.data_pb2 import DataRecord +from swanlab.proto.swanlab.metric.data.v1.scalar.scalar_pb2 import ScalarValue +from swanlab.proto.swanlab.record.v1.record_pb2 import Record + + +@pytest.fixture +def make_scalar_record(): + """工厂 fixture:创建一个 scalar 类型的 metric Record。""" + + def _make(step: int = 1) -> Record: + timestamp = Timestamp() + timestamp.GetCurrentTime() + return Record( + metric=DataRecord( + key="train/loss", + step=step, + timestamp=timestamp, + type=ColumnType.COLUMN_TYPE_FLOAT, + scalar=ScalarValue(number=0.125), + ) + ) + + return _make + + +@pytest.fixture +def make_config_record(): + """工厂 fixture:创建一个 config Record。""" + + def _make() -> Record: + timestamp = Timestamp() + timestamp.GetCurrentTime() + return Record(config=ConfigRecord(update_type=UpdateType.UPDATE_TYPE_PATCH, timestamp=timestamp)) + + return _make diff --git a/tests/unit/sdk/internal/core_python/uploader/test_collector.py b/tests/unit/sdk/internal/core_python/uploader/test_collector.py new file mode 100644 index 000000000..5cf4ed219 --- /dev/null +++ b/tests/unit/sdk/internal/core_python/uploader/test_collector.py @@ -0,0 +1,195 @@ +import threading +import time +from unittest.mock import MagicMock, patch + +from swanlab.sdk.internal.core_python.uploader.collector import Collector + +# ─────────────────── upload() ─────────────────── + + +def test_upload_delegates_to_upload_records(make_scalar_record): + """验证 upload() 将 pending 列表传递给 upload_records。""" + collector = Collector() + records = [make_scalar_record(step=1), make_scalar_record(step=2)] + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + collector.upload(records) + + mock_upload.assert_called_once_with(records, upload_callback=None) + + +def test_upload_passes_callback(make_scalar_record): + """验证 upload() 将构造时传入的 upload_callback 透传给 upload_records。""" + callback = MagicMock() + collector = Collector(upload_callback=callback) + records = [make_scalar_record(step=1)] + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + collector.upload(records) + + mock_upload.assert_called_once_with(records, upload_callback=callback) + + +def test_upload_skips_empty_list(): + """验证 upload() 传入空列表时不调用 upload_records。""" + collector = Collector() + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + collector.upload([]) + + mock_upload.assert_not_called() + + +# ─────────────────── submit() ─────────────────── + + +def test_submit_accumulates_and_uploads_when_interval_elapsed(make_scalar_record): + """验证 submit() 在超过 upload_interval 后聚合并上传累积的记录。""" + collector = Collector(upload_interval=0.0) + records = [make_scalar_record(step=i) for i in range(3)] + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + collector.submit(records) + + mock_upload.assert_called_once() + uploaded = mock_upload.call_args.args[0] + assert len(uploaded) == 3 + + +def test_submit_uploads_immediately_when_zero_interval_and_clock_does_not_advance(make_scalar_record): + """验证 upload_interval=0.0 时会立即上传。""" + records = [make_scalar_record(step=1)] + + with patch("swanlab.sdk.internal.core_python.uploader.collector.time.time", side_effect=[100.0, 100.0]): + collector = Collector(upload_interval=0.0) + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + collector.submit(records) + + mock_upload.assert_called_once() + uploaded = mock_upload.call_args.args[0] + assert len(uploaded) == 1 + + +def test_submit_accumulates_without_upload_when_interval_not_elapsed(make_scalar_record): + """验证 submit() 在未超过 upload_interval 时只聚合不上传。""" + collector = Collector(upload_interval=999.0) + records = [make_scalar_record(step=1)] + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + collector.submit(records) + + mock_upload.assert_not_called() + assert len(collector.container) == 1 + + +def test_submit_empty_records_is_noop(): + """验证 submit() 传入空列表时无任何副作用。""" + collector = Collector() + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + collector.submit([]) + + mock_upload.assert_not_called() + assert collector.container == [] + + +def test_submit_accumulates_across_calls_then_uploads(make_scalar_record): + """验证多次 submit() 调用累积记录,直到超过 interval 才一次性上传。""" + collector = Collector(upload_interval=0.0) + records_1 = [make_scalar_record(step=1)] + records_2 = [make_scalar_record(step=2)] + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + collector.submit(records_1) + collector.submit(records_2) + + assert mock_upload.call_count == 2 + # 第二次调用时 container 里只有 records_2,因为第一次已清空 + second_uploaded = mock_upload.call_args_list[1].args[0] + assert len(second_uploaded) == 1 + + +def test_submit_rolls_back_on_upload_error(make_scalar_record): + """验证 submit() 上传失败时,pending 记录被回滚到 container 头部。""" + collector = Collector(upload_interval=0.0) + # 先放入一条记录模拟已积累 + collector.container.append(make_scalar_record(step=0)) + new_records = [make_scalar_record(step=1)] + + with patch( + "swanlab.sdk.internal.core_python.uploader.collector.upload_records", side_effect=RuntimeError("network down") + ), patch("swanlab.sdk.internal.core_python.uploader.collector.console") as mock_console: + collector.submit(new_records) + + mock_console.error.assert_called_once() + assert "network down" in mock_console.error.call_args[0][0] + # 回滚后 container 应为 pending + 新增 = [old, new] + assert len(collector.container) == 2 + assert collector.container[0].metric.step == 0 + assert collector.container[1].metric.step == 1 + + +# ─────────────────── flush() ─────────────────── + + +def test_flush_flushes_all_accumulated_records(make_scalar_record): + """验证 flush() 无条件将所有累积记录一次性上传。""" + collector = Collector(upload_interval=999.0) + collector.container.append(make_scalar_record(step=1)) + + new_records = [make_scalar_record(step=2)] + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + collector.flush(new_records) + + mock_upload.assert_called_once() + uploaded = mock_upload.call_args.args[0] + assert len(uploaded) == 2 + assert collector.container == [] + + +def test_flush_empty_list_no_upload(): + """验证 flush() 传入空列表且 container 也为空时不触发上传。""" + collector = Collector() + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + collector.flush([]) + + mock_upload.assert_not_called() + + +def test_flush_rolls_back_on_upload_error(make_scalar_record): + """验证 flush() 上传失败时,pending 记录被回滚到 container。""" + collector = Collector() + records = [make_scalar_record(step=1), make_scalar_record(step=2)] + + with patch( + "swanlab.sdk.internal.core_python.uploader.collector.upload_records", side_effect=ValueError("bad data") + ), patch("swanlab.sdk.internal.core_python.uploader.collector.console") as mock_console: + collector.flush(records) + + mock_console.error.assert_called_once() + assert "bad data" in mock_console.error.call_args[0][0] + assert len(collector.container) == 2 + + +def test_flush_waits_for_lock_release(make_scalar_record): + """验证 flush() 在锁被持有时阻塞等待,释放后继续执行。""" + collector = Collector() + record = make_scalar_record(step=1) + done = threading.Event() + + collector._lock.acquire() + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + worker = threading.Thread(target=lambda: (collector.flush([record]), done.set())) + worker.start() + + time.sleep(0.05) + assert not done.is_set() + mock_upload.assert_not_called() + + collector._lock.release() + worker.join(timeout=1) + + assert done.is_set() diff --git a/tests/unit/sdk/internal/core_python/uploader/test_sender.py b/tests/unit/sdk/internal/core_python/uploader/test_sender.py new file mode 100644 index 000000000..e60e471ed --- /dev/null +++ b/tests/unit/sdk/internal/core_python/uploader/test_sender.py @@ -0,0 +1,160 @@ +from typing import Sequence, cast +from unittest.mock import MagicMock, call, patch + +from swanlab.proto.swanlab.record.v1.record_pb2 import Record +from swanlab.sdk.internal.core_python.uploader.sender import ( + HttpRecordTransport, + create_record_transport, + trace_records, + upload_records, +) + + +def test_http_record_transport_upload_record_group_skips_empty(): + """验证上传空列表时不抛异常,直接跳过。""" + transport = HttpRecordTransport() + transport.upload_record_group("metric", []) # should not raise + + +def test_http_record_transport_close(): + """验证 close() 不抛异常(当前为空实现)。""" + transport = HttpRecordTransport() + transport.close() # should not raise + + +# ─────────────────── create_record_transport ─────────────────── + + +def test_create_record_transport_returns_http_transport(): + """验证工厂函数返回 HttpRecordTransport 实例。""" + transport = create_record_transport() + + assert isinstance(transport, HttpRecordTransport) + + +# ─────────────────── trace_records ─────────────────── + + +def test_trace_records_groups_records_by_record_type(make_scalar_record, make_config_record): + """验证 trace_records 按 record_type 分组后分别上传,同类型 Record 合并为一次调用。""" + transport = MagicMock() + metric_1 = make_scalar_record(step=1) + config = make_config_record() + metric_2 = make_scalar_record(step=2) + records = [metric_1, config, metric_2] + + trace_records(records, per_request_len=-1, transport=transport) + + assert transport.upload_record_group.call_args_list == [ + call("metric", [metric_1, metric_2]), + call("config", [config]), + ] + transport.close.assert_not_called() + + +def test_trace_records_uploads_all_records_in_batches(make_scalar_record): + """验证超过 per_request_len 的记录被正确分片,每片独立上传并触发回调。""" + transport = MagicMock() + callback = MagicMock() + records = [make_scalar_record(step=index) for index in range(2500)] + + with patch( + "swanlab.sdk.internal.core_python.uploader.sender.create_record_transport", return_value=transport + ) as factory: + trace_records(records, per_request_len=1000, upload_callback=callback) + + assert transport.upload_record_group.call_args_list == [ + call("metric", records[:1000]), + call("metric", records[1000:2000]), + call("metric", records[2000:]), + ] + assert callback.call_args_list == [call(1000), call(1000), call(500)] + factory.assert_called_once_with() + transport.close.assert_called_once_with() + + +def test_trace_records_rejects_serialized_bytes(make_scalar_record): + """验证传入序列化字节而非 Record 实例时抛出 TypeError。""" + transport = MagicMock() + bad_records = cast(Sequence[Record], [make_scalar_record(step=7).SerializeToString()]) + + try: + trace_records(bad_records, transport=transport) + except TypeError as exc: + assert "Record" in str(exc) + else: + raise AssertionError("trace_records should reject serialized bytes") + + +def test_trace_records_none_input(): + """验证 records=None 时直接返回,不触发任何上传。""" + transport = MagicMock() + trace_records(records=None, transport=transport) + + transport.upload_record_group.assert_not_called() + + +def test_trace_records_empty_input(): + """验证空列表时直接返回,不触发任何上传。""" + transport = MagicMock() + trace_records(records=[], transport=transport) + + transport.upload_record_group.assert_not_called() + + +def test_trace_records_creates_and_closes_transport_when_none_provided(make_scalar_record): + """验证未提供 transport 时自动创建并在 finally 中关闭。""" + records = [make_scalar_record(step=1)] + + with patch("swanlab.sdk.internal.core_python.uploader.sender.create_record_transport") as factory: + mock_transport = MagicMock() + factory.return_value = mock_transport + + trace_records(records, per_request_len=-1) + + factory.assert_called_once_with() + mock_transport.upload_record_group.assert_called_once() + mock_transport.close.assert_called_once_with() + + +def test_trace_records_does_not_close_provided_transport(make_scalar_record): + """验证外部提供的 transport 在完成后不会被关闭。""" + transport = MagicMock() + records = [make_scalar_record(step=1)] + + trace_records(records, per_request_len=-1, transport=transport) + + transport.upload_record_group.assert_called_once() + transport.close.assert_not_called() + + +def test_trace_records_callback_none(make_scalar_record): + """验证 upload_callback=None 时不触发回调。""" + transport = MagicMock() + records = [make_scalar_record(step=1)] + + trace_records(records, per_request_len=-1, upload_callback=None, transport=transport) + + transport.upload_record_group.assert_called_once() + + +# ─────────────────── upload_records ─────────────────── + + +def test_upload_records_empty_list(): + """验证 upload_records 传入空列表时输出 debug 日志并返回。""" + with patch("swanlab.sdk.internal.core_python.uploader.sender.console") as mock_console: + upload_records([]) + + mock_console.debug.assert_called_once_with("No records to upload.") + + +def test_upload_records_delegates_to_trace_records(make_scalar_record): + """验证 upload_records 正确委托给 trace_records。""" + records = [make_scalar_record(step=1)] + callback = MagicMock() + + with patch("swanlab.sdk.internal.core_python.uploader.sender.trace_records") as mock_trace: + upload_records(records, upload_callback=callback, per_request_len=500) + + mock_trace.assert_called_once_with(records, per_request_len=500, upload_callback=callback) diff --git a/tests/unit/sdk/internal/core_python/uploader/test_thread.py b/tests/unit/sdk/internal/core_python/uploader/test_thread.py new file mode 100644 index 000000000..4ee95732c --- /dev/null +++ b/tests/unit/sdk/internal/core_python/uploader/test_thread.py @@ -0,0 +1,194 @@ +import time +from unittest.mock import MagicMock, patch + +from swanlab.sdk.internal.core_python.uploader.thread import Uploader + + +def test_uploader_defaults_to_class_upload_interval(): + """验证不传 upload_interval 时使用类默认值 UPLOAD_INTERVAL。""" + with patch("swanlab.sdk.internal.core_python.uploader.thread.Timer") as mock_timer_cls: + Uploader(auto_start=False) + + _, kwargs = mock_timer_cls.call_args + assert kwargs["interval"] == Uploader.UPLOAD_INTERVAL + + +def test_uploader_uses_custom_upload_interval(): + """验证传入自定义 upload_interval 时覆盖类默认值。""" + with patch("swanlab.sdk.internal.core_python.uploader.thread.Timer") as mock_timer_cls: + Uploader(upload_interval=0.5, auto_start=False) + + _, kwargs = mock_timer_cls.call_args + assert kwargs["interval"] == 0.5 + + +# ─────────────────── Uploader.start() ─────────────────── + + +def test_uploader_start_reuses_pkg_timer_scheduler(): + """验证 start() 创建 Timer 时参数正确(interval、immediate、name)并启动。""" + timer = MagicMock() + + with patch("swanlab.sdk.internal.core_python.uploader.thread.Timer") as mock_timer_cls: + mock_timer_cls.return_value = timer + + pool = Uploader(auto_start=False) + pool.start() + + assert mock_timer_cls.call_count == 1 + _, kwargs = mock_timer_cls.call_args + assert kwargs["interval"] == Uploader.UPLOAD_INTERVAL + assert kwargs["immediate"] is True + assert kwargs["name"] == Uploader.UPLOAD_THREAD_NAME + timer.start.assert_called_once_with() + + +def test_uploader_start_is_idempotent(): + """验证 start() 重复调用不会启动第二个 Timer。""" + timer = MagicMock() + + with patch("swanlab.sdk.internal.core_python.uploader.thread.Timer") as mock_timer_cls: + mock_timer_cls.return_value = timer + + pool = Uploader(auto_start=False) + pool.start() + pool.start() + + timer.start.assert_called_once_with() + + +def test_uploader_start_after_finish_is_noop(): + """验证 finish() 后再次调用 start() 不会重新启动。""" + pool = Uploader(auto_start=False) + pool.finish() + + with patch("swanlab.sdk.internal.core_python.uploader.thread.Timer") as mock_timer_cls: + pool.start() + + mock_timer_cls.assert_not_called() + + +# ─────────────────── Uploader.put() ─────────────────── + + +def test_uploader_put_adds_to_queue(make_config_record): + """验证 put() 将记录列表放入队列。""" + pool = Uploader(auto_start=False) + records = [make_config_record(), make_config_record()] + pool.put(records) + + assert pool._record_queue.qsize() == 1 + pool.finish() + + +def test_uploader_put_after_finish_raises(make_config_record): + """验证 finish() 后调用 put() 抛出 RuntimeError。""" + pool = Uploader(auto_start=False) + pool.finish() + + try: + pool.put([make_config_record()]) + raise AssertionError("put() should raise RuntimeError after finish") + except RuntimeError as exc: + assert "already been finished" in str(exc) + + +# ─────────────────── Uploader._drain_records() ─────────────────── + + +def test_uploader_drain_records_returns_pending_records_once(make_config_record): + """验证 _drain_records() 取出所有队列记录,再次调用返回空列表。""" + pool = Uploader(auto_start=False) + records = [make_config_record(), make_config_record()] + pool.put(records) + + assert pool._drain_records() == records + assert pool._drain_records() == [] + + +def test_uploader_drain_records_empty_queue(): + """验证队列为空时 _drain_records() 返回空列表。""" + pool = Uploader(auto_start=False) + + assert pool._drain_records() == [] + + +# ─────────────────── Uploader.finish() ─────────────────── + + +def test_uploader_finish_is_idempotent(make_config_record): + """验证 finish() 重复调用只执行一次上传和 Timer 清理。""" + pool = Uploader(auto_start=False) + pool.put([make_config_record(), make_config_record()]) + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + pool.finish() + pool.finish() + + assert mock_upload.call_count == 1 + uploaded = mock_upload.call_args.args[0] + assert len(uploaded) == 2 + assert all(record.WhichOneof("record_type") == "config" for record in uploaded) + + +def test_uploader_finish_cancels_and_joins_timer(): + """验证 finish() 取消并等待 Timer 线程退出。""" + timer = MagicMock() + + with patch("swanlab.sdk.internal.core_python.uploader.thread.Timer") as mock_timer_cls: + mock_timer_cls.return_value = timer + + pool = Uploader(auto_start=False) + pool.start() + pool.finish() + + timer.cancel.assert_called_once_with() + timer.join.assert_called_once_with(timeout=10) + + +def test_uploader_finish_drains_and_flushes_remaining(make_scalar_record): + """验证 finish() 排空队列中的残余记录并通过 flush 上传。""" + pool = Uploader(auto_start=False) + pool.put([make_scalar_record(step=1), make_scalar_record(step=2)]) + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + pool.finish() + + mock_upload.assert_called_once() + uploaded = mock_upload.call_args.args[0] + assert len(uploaded) == 2 + + +def test_uploader_finish_passes_upload_callback(make_scalar_record): + """验证 finish() 通过 Collector 将 upload_callback 透传到上传层。""" + callback = MagicMock() + pool = Uploader(upload_interval=0.5, upload_callback=callback, auto_start=False) + pool.put([make_scalar_record(step=1)]) + + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + pool.finish() + + mock_upload.assert_called_once() + # upload_records 第二个参数应为 upload_callback + assert mock_upload.call_args.kwargs.get("upload_callback") is callback + + +# ─────────────────── Uploader 集成测试 ─────────────────── + + +def test_uploader_starts_upload_thread_automatically(make_scalar_record): + """验证 auto_start=True 时自动启动 Timer,put 的记录能在超时后被上传。""" + with patch.object(Uploader, "UPLOAD_INTERVAL", 0.01): + pool = Uploader(upload_interval=0.01) + + try: + with patch("swanlab.sdk.internal.core_python.uploader.collector.upload_records") as mock_upload: + pool.put([make_scalar_record()]) + + deadline = time.time() + 1.0 + while mock_upload.call_count == 0 and time.time() < deadline: + time.sleep(0.02) + + assert mock_upload.call_count == 1 + finally: + pool.finish() diff --git a/tests/unit/sdk/internal/core_python/uploader/test_uploader_helper.py b/tests/unit/sdk/internal/core_python/uploader/test_uploader_helper.py new file mode 100644 index 000000000..4aef78a9d --- /dev/null +++ b/tests/unit/sdk/internal/core_python/uploader/test_uploader_helper.py @@ -0,0 +1,111 @@ +import pytest + +from swanlab.proto.swanlab.record.v1.record_pb2 import Record +from swanlab.sdk.internal.core_python.uploader.helper import generate_chunks, group_records_by_type + +# ─────────────────── generate_chunks ─────────────────── + + +def test_generate_chunks_single_chunk(make_scalar_record): + """验证记录数不超过 per_request_len 时产出单个分片。""" + records = [make_scalar_record(step=i) for i in range(5)] + chunks = list(generate_chunks(records, per_request_len=10)) + + assert len(chunks) == 1 + assert chunks[0] == (records, 5) + + +def test_generate_chunks_exact_boundary(make_scalar_record): + """验证记录数恰好等于 per_request_len 时产出单个分片,不拆分。""" + records = [make_scalar_record(step=i) for i in range(100)] + chunks = list(generate_chunks(records, per_request_len=100)) + + assert len(chunks) == 1 + assert chunks[0] == (records, 100) + + +def test_generate_chunks_multiple_chunks(make_scalar_record): + """验证记录数超过 per_request_len 时正确拆分为多个分片,最后一片长度不足也正常产出。""" + records = [make_scalar_record(step=i) for i in range(25)] + chunks = list(generate_chunks(records, per_request_len=10)) + + assert len(chunks) == 3 + assert chunks[0][1] == 10 + assert chunks[1][1] == 10 + assert chunks[2][1] == 5 + + +def test_generate_chunks_no_limit(make_scalar_record): + """验证 per_request_len=-1 时不拆分,一次性返回全部记录。""" + records = [make_scalar_record(step=i) for i in range(9999)] + chunks = list(generate_chunks(records, per_request_len=-1)) + + assert len(chunks) == 1 + assert chunks[0] == (records, 9999) + + +def test_generate_chunks_invalid_per_request_len(make_scalar_record): + """验证 per_request_len 为 0 或负数(非 -1)时抛出 ValueError。""" + records = [make_scalar_record(step=1)] + + with pytest.raises(ValueError, match="per_request_len"): + list(generate_chunks(records, per_request_len=0)) + + with pytest.raises(ValueError, match="per_request_len"): + list(generate_chunks(records, per_request_len=-5)) + + +def test_generate_chunks_empty_records(): + """验证空记录列表在 per_request_len > 0 时产出单个空分片。""" + chunks = list(generate_chunks([], per_request_len=10)) + assert len(chunks) == 1 + assert chunks[0] == ([], 0) + + +def test_generate_chunks_no_limit_empty(): + """验证 per_request_len=-1 且空列表时产出单个空分片。""" + chunks = list(generate_chunks([], per_request_len=-1)) + assert len(chunks) == 1 + assert chunks[0] == ([], 0) + + +# ─────────────────── group_records_by_type ─────────────────── + + +def test_group_records_by_type_basic(make_scalar_record, make_config_record): + """验证不同 record_type 的 Record 被正确分组到各自的列表中。""" + metric_1 = make_scalar_record(step=1) + config = make_config_record() + metric_2 = make_scalar_record(step=2) + + grouped = group_records_by_type([metric_1, config, metric_2]) + + assert set(grouped.keys()) == {"metric", "config"} + assert grouped["metric"] == [metric_1, metric_2] + assert grouped["config"] == [config] + + +def test_group_records_by_type_preserves_order(make_scalar_record, make_config_record): + """验证分组的 key 按 Record 首次出现顺序排列(OrderedDict)。""" + config = make_config_record() + metric = make_scalar_record(step=1) + + grouped = group_records_by_type([config, metric]) + assert list(grouped.keys()) == ["config", "metric"] + + +def test_group_records_by_type_empty(): + """验证空列表返回空字典。""" + assert group_records_by_type([]) == {} + + +def test_group_records_by_type_rejects_non_record(): + """验证传入非 Record 实例时抛出 TypeError。""" + with pytest.raises(TypeError, match="Record"): + group_records_by_type(["not_a_record"]) # type: ignore[list-item] + + +def test_group_records_by_type_rejects_empty_record(): + """验证未设置任何 record_type 的空 Record 抛出 ValueError。""" + with pytest.raises(ValueError, match="no active record_type"): + group_records_by_type([Record()]) diff --git a/tests/unit/sdk/internal/pkg/timer/test_timer.py b/tests/unit/sdk/internal/pkg/timer/test_timer.py new file mode 100644 index 000000000..775223170 --- /dev/null +++ b/tests/unit/sdk/internal/pkg/timer/test_timer.py @@ -0,0 +1,207 @@ +import threading +import time +from unittest.mock import MagicMock + +import pytest + +from swanlab.sdk.internal.pkg import console +from swanlab.sdk.internal.pkg.timer import Timer + + +@pytest.fixture +def timer_manager(): + timers = [] + + def _create_timer(*args, **kwargs): + timer = Timer(*args, **kwargs) + timers.append(timer) + return timer + + yield _create_timer + + for timer in timers: + try: + timer.cancel() + timer.join(timeout=1.0) + except Exception: + pass + + +class TestTimer: + def test_basic_interval(self, timer_manager): + """验证固定间隔下任务被周期执行,且 execution_count 正确递增。""" + task_triggered = threading.Event() + + def task(): + if timer.execution_count >= 1: + task_triggered.set() + + timer = timer_manager(task, interval=0.1, immediate=False) + timer.run() + + assert task_triggered.wait(timeout=2.0), "Task was not executed twice" + + timer.cancel() + timer.join(timeout=1.0) + + assert timer.execution_count >= 2 + assert not timer.is_running + + def test_immediate_execution(self, timer_manager): + """验证 immediate=True 时任务在启动后立即执行一次,不计入后续等待周期。""" + task_triggered = threading.Event() + + timer = timer_manager(task_triggered.set, interval=10.0, immediate=True) + timer.run() + + assert task_triggered.wait(2.0) + + timer.cancel() + timer.join(timeout=1.0) + + assert timer.execution_count == 1 + + def test_dynamic_interval(self, timer_manager): + """验证动态间隔策略接收已执行次数并据此调整间隔,策略入参递增。""" + reached_slow_interval = threading.Event() + interval_inputs = [] + + def interval_strategy(count: int) -> float: + interval_inputs.append(count) + if count < 2: + return 0.1 + reached_slow_interval.set() + return 0.5 + + timer = timer_manager(lambda: None, interval=interval_strategy) + timer.run() + + assert reached_slow_interval.wait(2.0) + + timer.cancel() + timer.join(timeout=1.0) + + assert interval_inputs[:3] == [0, 1, 2] + assert timer.execution_count >= 2 + + def test_cancel_interrupts_sleep(self, timer_manager): + """验证 cancel() 能立即唤醒正在等待间隔的线程,不会等到间隔结束。""" + start_time = time.monotonic() + + timer = timer_manager(lambda: None, interval=10.0) + timer.run() + + time.sleep(0.1) + + timer.cancel() + timer.join(timeout=1.0) + + assert time.monotonic() - start_time < 1.0 + assert not timer.is_running + + def test_error_resilience(self, timer_manager, monkeypatch): + """验证任务抛异常不影响后续轮次执行,异常信息通过 console.error 输出。""" + error_mock = MagicMock() + monkeypatch.setattr(console, "error", error_mock) + + third_call = threading.Event() + + def failing_task() -> None: + if timer.execution_count >= 2: + third_call.set() + raise ValueError("Test Error") + + timer = timer_manager(failing_task, interval=0.1) + timer.run() + + assert third_call.wait(2.0) + + timer.cancel() + timer.join(timeout=1.0) + + assert timer.execution_count >= 3 + assert "Error executing task" in error_mock.call_args[0][0] + + def test_double_run_warning(self, timer_manager, monkeypatch): + """验证运行中再次调用 run() 只告警不启动重复线程。""" + warning_mock = MagicMock() + monkeypatch.setattr(console, "warning", warning_mock) + + task_triggered = threading.Event() + timer = timer_manager(task_triggered.set, interval=1.0, immediate=True) + timer.run() + + assert task_triggered.wait(2.0) + + timer.run() + + warning_mock.assert_called_once_with("Timer already running") + + timer.cancel() + timer.join(timeout=1.0) + + def test_restart_capability(self, timer_manager): + """验证 cancel + join 后可以重新 run(),execution_count 持续累加。""" + task_triggered = threading.Event() + + timer = timer_manager(task_triggered.set, interval=0.1) + + timer.run() + assert task_triggered.wait(timeout=2.0), "First run: task was never executed" + timer.cancel() + timer.join(timeout=1.0) + + first_run_count = timer.execution_count + assert first_run_count >= 1 + + task_triggered.clear() + timer.run() + assert task_triggered.wait(timeout=2.0), "Second run: task was never executed" + timer.cancel() + timer.join(timeout=1.0) + + assert timer.execution_count > first_run_count + + @pytest.mark.parametrize("interval", [0, -1, False]) + def test_reject_invalid_fixed_interval(self, interval): + """验证构造时传入 0、负数或 bool 作为间隔会抛出 TypeError/ValueError。""" + with pytest.raises((TypeError, ValueError), match="Timer interval"): + Timer(lambda: None, interval=interval) + + def test_interval_strategy_runtime_error(self, timer_manager, monkeypatch): + """验证动态间隔策略在运行时抛异常时,循环终止并输出错误日志。""" + error_mock = MagicMock() + monkeypatch.setattr(console, "error", error_mock) + + call_count = 0 + + def failing_strategy(count: int) -> float: + nonlocal call_count + call_count += 1 + if call_count >= 2: + raise RuntimeError("strategy exploded") + return 0.05 + + timer = timer_manager(lambda: None, interval=failing_strategy) + timer.run() + + # 等待循环因策略异常而退出 + deadline = time.monotonic() + 3.0 + while timer.is_running and time.monotonic() < deadline: + time.sleep(0.05) + + assert not timer.is_running, "Timer should have stopped after interval strategy error" + assert "Timer interval strategy error" in error_mock.call_args[0][0] + + def test_cancel_before_start_has_no_effect(self, timer_manager): + """验证在启动前调用 cancel() 不会阻止后续启动(start() 会重置 stop_event)。""" + task_triggered = threading.Event() + + timer = timer_manager(task_triggered.set, interval=1.0, immediate=True) + timer.cancel() + timer.run() + + assert task_triggered.wait(timeout=2.0), "Timer should start normally despite prior cancel" + timer.cancel() + timer.join(timeout=1.0) + assert timer.execution_count >= 1 diff --git a/tests/unit/sdk/utils/helper/test_helper.py b/tests/unit/sdk/utils/helper/test_util_helper.py similarity index 100% rename from tests/unit/sdk/utils/helper/test_helper.py rename to tests/unit/sdk/utils/helper/test_util_helper.py