diff --git a/swanlab/sdk/cmd/login.py b/swanlab/sdk/cmd/login.py index 77ab5f802..e1020edab 100644 --- a/swanlab/sdk/cmd/login.py +++ b/swanlab/sdk/cmd/login.py @@ -12,9 +12,10 @@ from swanlab.sdk.internal import apikey from swanlab.sdk.internal.context import RunConfig, RunContext, get_context, has_context, use_temp_context from swanlab.sdk.internal.core_python import client +from swanlab.sdk.internal.pkg import console from swanlab.sdk.internal.pkg.scope import Scope from swanlab.sdk.internal.settings import Settings, settings -from swanlab.sdk.pkg import console, helper +from swanlab.sdk.pkg import helper from swanlab.sdk.typings.core_python.api.bootstrap import LoginResponse diff --git a/swanlab/sdk/internal/apikey/__init__.py b/swanlab/sdk/internal/apikey/__init__.py index 3585f4458..c37d980d7 100644 --- a/swanlab/sdk/internal/apikey/__init__.py +++ b/swanlab/sdk/internal/apikey/__init__.py @@ -11,9 +11,10 @@ from rich.text import Text +from swanlab.sdk.internal.pkg import console from swanlab.sdk.internal.pkg.netrc import get_nrc_path, remove_host_suffix, write_netrc from swanlab.sdk.internal.pkg.settings import get_current_settings -from swanlab.sdk.pkg import console, helper +from swanlab.sdk.pkg import helper __all__ = ["get", "save", "exists"] diff --git a/swanlab/sdk/internal/core_python/client/__init__.py b/swanlab/sdk/internal/core_python/client/__init__.py index 35a89eba8..9edec96cd 100644 --- a/swanlab/sdk/internal/core_python/client/__init__.py +++ b/swanlab/sdk/internal/core_python/client/__init__.py @@ -14,10 +14,10 @@ from swanlab.sdk.internal.core_python.api.bootstrap import login_by_api_key from swanlab.sdk.internal.pkg.netrc import remove_host_suffix from swanlab.sdk.internal.pkg.scope import set_context -from swanlab.sdk.pkg import console from swanlab.sdk.pkg.version import get_swanlab_version from swanlab.sdk.typings.core_python.api.bootstrap import LoginResponse +from ...pkg import console from . import session from .helper import decode_response diff --git a/swanlab/sdk/internal/core_python/client/session.py b/swanlab/sdk/internal/core_python/client/session.py index 14d858f31..542d7f922 100644 --- a/swanlab/sdk/internal/core_python/client/session.py +++ b/swanlab/sdk/internal/core_python/client/session.py @@ -8,7 +8,8 @@ import contextvars import copy -from typing import Optional +import time +from typing import Optional, Union from requests import Session from requests.adapters import HTTPAdapter @@ -16,12 +17,14 @@ from swanlab.exceptions import ApiError from swanlab.sdk.internal.core_python.client.helper import decode_error_response +from swanlab.sdk.internal.pkg import log +from swanlab.sdk.pkg import helper from swanlab.sdk.pkg.version import get_swanlab_version __all__ = ["create", "TimeoutHTTPAdapter", "SessionWithRetry"] VERSION_HEADER = "X-SwanLab-SDK-Version" # 用于存储当前请求的重试次数,避免在请求中传递 retries 参数 -request_retries_ctx = contextvars.ContextVar("request_retries", default=None) +request_retries_ctx: contextvars.ContextVar[Optional[int]] = contextvars.ContextVar("request_retries", default=None) class TimeoutHTTPAdapter(HTTPAdapter): @@ -51,6 +54,25 @@ def send(self, request, *args, **kwargs): return super().send(request, *args, **kwargs) +@helper.catch_and_return_none() +def format_body_preview(body: Optional[Union[bytes, str]], max_len: int = 1000) -> str: + """ + 统一格式化并截断请求/响应体,安全处理二进制乱码。 + 如果是 bytes,尝试用 utf-8 解码;超过 max_len 则截断防刷屏。 + """ + if not body: + return "" + + if isinstance(body, bytes): + body_str = body.decode("utf-8", errors="replace") + else: + body_str = str(body) + + if len(body_str) > max_len: + return body_str[:max_len] + " ... (truncated)" + return body_str + + class SessionWithRetry(Session): """ 支持在请求级别自定义重试次数的 Session。 @@ -73,18 +95,52 @@ def request(self, method, url, *args, **kwargs): def send(self, request, **kwargs): """ - 重写底层发送方法,统一处理所有响应的校验逻辑 + 重写底层发送方法,统一处理所有响应的校验逻辑和网络日志记录 """ + method = (request.method or "unknown").upper() + + # --- [DEBUG] 记录请求详情 --- + if helper.env.DEBUG: + # 这里的 request.url 已经包含了 params 拼接后的完整 query 字符串 + log.debug("[HTTP-REQ] %s %s | Headers: %s", method, request.url, request.headers) + body = request.body + if body: + body_preview = format_body_preview(body) or "" + log.debug("[HTTP-REQ-BODY] %s", body_preview) + # --------------------------- + + start = time.perf_counter() + # 调用父类(或 Adapter)获取响应 response = super().send(request, **kwargs) - # 1. 2xx 响应直接放行 + elapsed_ms = (time.perf_counter() - start) * 1000 + trace_id = response.headers.get("traceid", "unknown") + + # 1. 2xx 响应:记录正常日志后放行 if response.ok: + log.debug( + "[HTTP] %s %s -> %s (%.0fms) trace:%s", + method, + request.url, + response.status_code, + elapsed_ms, + trace_id, + ) + + # --- [DEBUG] 记录成功响应详情 --- + if helper.env.DEBUG: + log.debug("[HTTP-RES] Headers: %s", response.headers) + # 直接传 response.text (str) 给格式化函数 + text = response.text + if text: + resp_preview = format_body_preview(text) or "" + log.debug("[HTTP-RES-BODY] %s", resp_preview) + # ------------------------------- + return response - # 2. 准备 Fallback 默认值 - method = (request.method or "unknown").upper() - trace_id = response.headers.get("traceid", "unknown") + # 2. 非 2xx 响应:准备 Fallback 默认值 error_code = "unknown code" error_message = "unknown error" @@ -93,7 +149,29 @@ def send(self, request, **kwargs): if decoded is not None: error_code, error_message = decoded - # 4. 抛出友好的自定义 ApiError + # 4. 记录错误日志(附带响应体,方便排查) + log.error( + "[HTTP] %s %s -> %s (%.0fms) trace:%s | [ERR] code=%s message=%s", + method, + request.url, + response.status_code, + elapsed_ms, + trace_id, + error_code, + error_message, + ) + + # --- [DEBUG] 记录失败响应详情 --- + if helper.env.DEBUG: + log.debug("[HTTP-RES-ERR] Headers: %s", response.headers) + text = response.text + if text and not decoded: + # 只有当解码失败时,才额外把原始错误 body 打印出来 + err_preview = format_body_preview(text) or "" + log.debug("[HTTP-RES-ERR-BODY] %s", err_preview) + # ------------------------------- + + # 5. 抛出友好的自定义 ApiError raise ApiError(response, method=method, trace_id=trace_id, code=error_code, message=error_message) # ---------------------------------- 类型提示占位符,保留以保证 IDE 友好 ---------------------------------- diff --git a/swanlab/sdk/pkg/console/__init__.py b/swanlab/sdk/internal/pkg/console/__init__.py similarity index 72% rename from swanlab/sdk/pkg/console/__init__.py rename to swanlab/sdk/internal/pkg/console/__init__.py index dd20be79b..b21ace85e 100644 --- a/swanlab/sdk/pkg/console/__init__.py +++ b/swanlab/sdk/internal/pkg/console/__init__.py @@ -3,17 +3,17 @@ @file: __init__.py @time: 2026/3/6 13:20 @description: SwanLab SDK 控制台日志模块,负责打印日志 +当 SWANLAB_DEBUG=true 时,终端输出会同步镜像到诊断日志文件(通过 log 模块) """ -import os +from typing import Any from rich.console import Console from rich.markup import escape from rich.text import Text -# 设计上 console 独立于其他模块,包括settings。但是我们又希望有一个环境变量去控制是否打印debug信息,所以这里额外绑定一个环境变量 -DEBUG = os.getenv("SWANLAB_DEBUG", "false").lower() in ["true", "1", "yes", "on"] - +from swanlab.sdk.internal.pkg import log +from swanlab.sdk.pkg import helper __all__ = ["debug", "info", "warning", "error", "critical", "disable_log", "enable_log"] @@ -62,6 +62,20 @@ def _print_formatted(level_name: str, *args, **kwargs): _console.print(prefix_text, *safe_args, **kwargs) +def _to_plain_text(*args: Any) -> str: + """ + 将 console 的 *args 转为纯文本字符串,用于写入诊断日志文件。 + 去除 rich markup,保留原始语义。 + """ + parts = [] + for arg in args: + if isinstance(arg, Text): + parts.append(arg.plain) + else: + parts.append(str(arg)) + return " ".join(parts) + + # ----------------------------------------------------------------------------- # 导出的模块级日志函数 # ----------------------------------------------------------------------------- @@ -77,9 +91,10 @@ def print(*args, **kwargs): # noqa: A001 def debug(*args, **kwargs): """发送调试消息""" - if not _can_log or not DEBUG: + if not _can_log or not helper.env.DEBUG: return _print_formatted("debug", *args, **kwargs) + log.debug("[CONSOLE] %s", _to_plain_text(*args)) def info(*args, **kwargs): @@ -87,6 +102,8 @@ def info(*args, **kwargs): if not _can_log: return _print_formatted("info", *args, **kwargs) + if helper.env.DEBUG: + log.info("[CONSOLE] %s", _to_plain_text(*args)) def warning(*args, **kwargs): @@ -94,6 +111,8 @@ def warning(*args, **kwargs): if not _can_log: return _print_formatted("warning", *args, **kwargs) + if helper.env.DEBUG: + log.warning("[CONSOLE] %s", _to_plain_text(*args)) def error(*args, **kwargs): @@ -101,6 +120,8 @@ def error(*args, **kwargs): if not _can_log: return _print_formatted("error", *args, **kwargs) + if helper.env.DEBUG: + log.error("[CONSOLE] %s", _to_plain_text(*args)) def critical(*args, **kwargs): @@ -108,3 +129,5 @@ def critical(*args, **kwargs): if not _can_log: return _print_formatted("critical", *args, **kwargs) + if helper.env.DEBUG: + log.critical("[CONSOLE] %s", _to_plain_text(*args)) diff --git a/swanlab/sdk/internal/pkg/log/__init__.py b/swanlab/sdk/internal/pkg/log/__init__.py new file mode 100644 index 000000000..2d70f6329 --- /dev/null +++ b/swanlab/sdk/internal/pkg/log/__init__.py @@ -0,0 +1,186 @@ +""" +@author: cunyue +@file: __init__.py +@time: 2026/3/10 14:00 +@description: SwanLab SDK 内部诊断日志模块 + +与 console 模块互补: +- console:面向用户的终端美化输出(rich) +- log:面向开发者/运维的文件持久化诊断日志(logging) + +同时 console 模块在 SWANLAB_DEBUG=true 时会将终端输出同步到诊断日志文件。 + +设计要点: +1. 基于 Python 标准库 logging,零额外依赖 +2. 使用独立命名空间 "swanlab.internal",不污染用户的 logging 配置 +3. 延迟绑定(Lazy Attach):log_dir 就绪前日志缓冲在内存中,就绪后一次性 flush 到文件 +4. RotatingFileHandler 自动轮转,防止日志文件无限膨胀 +""" + +import logging +import os +from logging.handlers import MemoryHandler, RotatingFileHandler +from pathlib import Path +from typing import Optional + +__all__ = ["debug", "info", "warning", "error", "critical", "bindfile", "reset"] + +# --------------------------------------------------------------------------- +# Logger 初始化 +# --------------------------------------------------------------------------- +_LOGGER_NAME = "swanlab.internal" +_logger = logging.getLogger(_LOGGER_NAME) +# 阻止日志向上传播到 root logger,避免干扰用户的 logging 配置 +_logger.propagate = False +# 内部诊断日志默认捕获 DEBUG 及以上级别的所有信息 +_logger.setLevel(logging.DEBUG) + +# 日志格式:时间 | 级别 | 来源 | 消息 +_FORMATTER = logging.Formatter( + fmt="%(asctime)s | %(levelname)-7s | %(name)s | %(message)s", + datefmt="%Y-%m-%dT%H:%M:%S", +) + +# --------------------------------------------------------------------------- +# 内存缓冲 Handler(阶段 1:log_dir 就绪前) +# capacity 设为 1024 条,足够覆盖 login 阶段的少量请求 +# flushLevel 设为 CRITICAL+1(即永远不会自动 flush),完全由 bindfile() 手动触发 +# --------------------------------------------------------------------------- +_memory_handler: Optional[MemoryHandler] = None +_file_handler: Optional[RotatingFileHandler] = None +_bound = False + +# --------------------------------------------------------------------------- +# 轮转配置常量 +# --------------------------------------------------------------------------- +_MAX_BYTES = 10 * 1024 * 1024 # 单文件上限 10 MB +_BACKUP_COUNT = 3 # 保留 3 个备份(共 ~40 MB) +_LOG_FILENAME = "debug.log" + + +class SecureRotatingFileHandler(RotatingFileHandler): + """ + 安全的日志轮转处理器。 + 确保每次创建/打开日志文件时,文件权限都被强制设置为 0600 (仅属主可读写)。 + 主要用于防护在 Linux/macOS 共享算力集群上的敏感凭证泄露。 + """ + + def _open(self): + stream = super()._open() + + # 仅在类 Unix 系统(Linux/macOS)上应用 POSIX 权限 + if os.name == "posix": + os.chmod(self.baseFilename, 0o600) + + # 注意:Windows 环境下多为个人开发机,且 os.chmod 无法操作 ACL,故跳过 + return stream + + +def reset() -> None: + """ + 重置日志模块状态,恢复到初始的内存缓冲模式。 + 释放现有的文件句柄,清除所有已绑定的 Handler。 + """ + global _memory_handler, _file_handler, _bound + + # 1. 安全关闭并移除所有现有的 Handlers(防止文件句柄泄露) + for handler in _logger.handlers[:]: + handler.close() + _logger.removeHandler(handler) + + # 2. 重新初始化内存缓冲 Handler + _memory_handler = MemoryHandler( + capacity=1024, + flushLevel=logging.CRITICAL + 1, + ) + _memory_handler.setFormatter(_FORMATTER) + _logger.addHandler(_memory_handler) + + # 3. 重置状态标志 + _file_handler = None + _bound = False + + +# 模块首次导入时,执行一次初始化 +reset() + + +def bindfile(log_dir: Path) -> None: + """ + 将诊断日志绑定到文件。应在 log_dir 目录创建后调用。 + + 执行流程: + 1. 创建 SecureRotatingFileHandler 指向 log_dir/debug.log + 2. 将内存缓冲中的日志 flush 到文件 + 3. 移除内存缓冲 Handler,后续日志直接写文件 + + 此方法可安全地重复调用(幂等),第二次调用会被忽略。 + + :param log_dir: 日志目录路径,调用方需确保目录已创建 + :raises FileNotFoundError: 如果 log_dir 不存在 + """ + global _memory_handler, _file_handler, _bound + + if _bound: + return + + log_dir = Path(log_dir) + if not log_dir.exists(): + raise FileNotFoundError(f"Log directory does not exist: {log_dir}") + + # 1. 创建文件 Handler + log_path = log_dir / _LOG_FILENAME + _file_handler = SecureRotatingFileHandler( + filename=str(log_path), + maxBytes=_MAX_BYTES, + backupCount=_BACKUP_COUNT, + encoding="utf-8", + ) + _file_handler.setFormatter(_FORMATTER) + _file_handler.setLevel(logging.DEBUG) + + # 2. 将内存缓冲的日志 flush 到文件 + if _memory_handler is not None: + _memory_handler.setTarget(_file_handler) + _memory_handler.flush() + # 移除内存 Handler,释放缓冲区 + _logger.removeHandler(_memory_handler) + _memory_handler.close() + _memory_handler = None + + # 3. 挂载文件 Handler + _logger.addHandler(_file_handler) + _bound = True + + _logger.debug("Diagnostic log bound to file: %s", log_path) + + +# --------------------------------------------------------------------------- +# 模块级日志函数 +# 与 console 模块保持相同的 API 风格,方便调用方统一使用 +# --------------------------------------------------------------------------- + + +def debug(msg: str, *args) -> None: + """记录调试级别的诊断日志""" + _logger.debug(msg, *args) + + +def info(msg: str, *args) -> None: + """记录信息级别的诊断日志""" + _logger.info(msg, *args) + + +def warning(msg: str, *args) -> None: + """记录警告级别的诊断日志""" + _logger.warning(msg, *args) + + +def error(msg: str, *args) -> None: + """记录错误级别的诊断日志""" + _logger.error(msg, *args) + + +def critical(msg: str, *args) -> None: + """记录致命级别的诊断日志""" + _logger.critical(msg, *args) diff --git a/swanlab/sdk/internal/settings/__init__.py b/swanlab/sdk/internal/settings/__init__.py index 8a5c0fa28..c1d26acd1 100644 --- a/swanlab/sdk/internal/settings/__init__.py +++ b/swanlab/sdk/internal/settings/__init__.py @@ -34,9 +34,9 @@ YamlConfigSettingsSource, ) -from swanlab.sdk.pkg import console from swanlab.sdk.typings.run import ModeType +from ..pkg import console from .experiment import ExperimentSettings, ProjectSettings, RunSettings from .integration import IntegrationSettings from .metadata import ConsoleSettings, EnvSettings, HardwareSettings diff --git a/swanlab/sdk/pkg/helper/env.py b/swanlab/sdk/pkg/helper/env.py index 02b6beaca..2a59634e9 100644 --- a/swanlab/sdk/pkg/helper/env.py +++ b/swanlab/sdk/pkg/helper/env.py @@ -11,6 +11,12 @@ __all__ = ["is_jupyter", "is_interactive"] +# 设计上,DEBUG 变量独立于其他模块,包括 settings。但是我们又希望有一个环境变量去控制是否打印 debug 信息,所以这里额外绑定一个环境变量 +DEBUG = os.getenv("SWANLAB_DEBUG", "false").lower() in ["true", "1", "yes", "on"] +""" +是否为调试模式 +""" + def is_jupyter() -> bool: """判断当前是否在 Jupyter Notebook 环境中""" diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index dff6820f1..cd80a448a 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -2,6 +2,7 @@ from swanlab.sdk.internal.context import clear_context from swanlab.sdk.internal.core_python import client +from swanlab.sdk.internal.pkg import log from swanlab.sdk.internal.settings import Settings, settings @@ -50,6 +51,10 @@ def isolate_sdk_environment(tmp_path, monkeypatch): # --- 4. 手动 Teardown 清理 --- # 确保测试结束后没有任何残留进入下一个用例 + # 1. 清理 Client 单例 if client.exists(): client.reset() + # 2. 清理 RunContext clear_context() + # 3. 清理 logger + log.reset() diff --git a/tests/unit/sdk/internal/core_python/client/test_client_session.py b/tests/unit/sdk/internal/core_python/client/test_client_session.py index 3f75a320a..061a91974 100644 --- a/tests/unit/sdk/internal/core_python/client/test_client_session.py +++ b/tests/unit/sdk/internal/core_python/client/test_client_session.py @@ -5,11 +5,18 @@ @description: 测试 SwanLab 运行时客户端会话辅助函数 """ +from unittest.mock import MagicMock + import pytest import responses from swanlab.exceptions import ApiError -from swanlab.sdk.internal.core_python.client.session import TimeoutHTTPAdapter, create, request_retries_ctx +from swanlab.sdk.internal.core_python.client.session import ( + TimeoutHTTPAdapter, + create, + format_body_preview, + request_retries_ctx, +) @pytest.fixture @@ -99,3 +106,121 @@ def test_request_retries_context_cleanup(session): # 无论中间发生了什么(包括抛出异常),finally 块必须确保清理了上下文 assert request_retries_ctx.get() is None + + +# ============================================================================== +# Helper Function: format_body_preview 测试 +# ============================================================================== + + +def test_format_body_preview(): + """测试格式化截断函数的各项能力""" + # 1. 空值处理 + assert format_body_preview(None) == "" + assert format_body_preview(b"") == "" + assert format_body_preview("") == "" + + # 2. 基础字符串及未越界截断 + assert format_body_preview("hello world", max_len=20) == "hello world" + assert format_body_preview("hello world", max_len=5) == "hello ... (truncated)" + + # 3. 正常 utf-8 bytes 解码 + assert format_body_preview(b'{"key":"value"}') == '{"key":"value"}' + + # 4. 无法正常解码的乱码 bytes(预期会被 errors="replace" 替换为占位符,不会抛错) + bad_bytes = b"\xff\xfe\xfd" + result = format_body_preview(bad_bytes) + assert result is not None + assert "\ufffd" in result # \ufffd 就是那个长得像问号的替换字符 + + +# ============================================================================== +# DEBUG 日志相关测试 +# ============================================================================== + + +@responses.activate() +def test_session_debug_logging_success(session, monkeypatch): + """测试开启 DEBUG 时,成功请求会记录详细的请求/响应体和 Headers""" + monkeypatch.setattr("swanlab.sdk.internal.core_python.client.session.helper.env.DEBUG", True) + mock_debug = MagicMock() + monkeypatch.setattr("swanlab.sdk.internal.core_python.client.session.log.debug", mock_debug) + + responses.add( + responses.POST, + "http://api.test/debug-success", + json={"status": "ok"}, + status=200, + ) + + session.post("http://api.test/debug-success", json={"req_key": "req_val"}) + + all_logged_args = [str(arg) for call in mock_debug.call_args_list for arg in call.args] + + # 验证格式化字符串标签存在 + assert any("[HTTP-REQ]" in msg for msg in all_logged_args) + assert any("[HTTP-REQ-BODY] %s" in msg for msg in all_logged_args) + assert any("[HTTP-RES]" in msg for msg in all_logged_args) + assert any("[HTTP-RES-BODY] %s" in msg for msg in all_logged_args) + + # 验证实际的请求/响应体被解码并当做变量传入 + assert any('{"req_key": "req_val"}' in msg for msg in all_logged_args) + assert any('{"status": "ok"}' in msg for msg in all_logged_args) + + +@responses.activate() +def test_session_debug_logging_error(session, monkeypatch): + """测试开启 DEBUG 时,失败请求会记录详细的响应 Headers 和原始 Body""" + monkeypatch.setattr("swanlab.sdk.internal.core_python.client.session.helper.env.DEBUG", True) + mock_debug = MagicMock() + monkeypatch.setattr("swanlab.sdk.internal.core_python.client.session.log.debug", mock_debug) + + responses.add( + responses.GET, + "http://api.test/debug-error", + body="Nginx 502 Bad Gateway", + status=502, + ) + + with pytest.raises(ApiError): + session.get("http://api.test/debug-error") + + all_logged_args = [str(arg) for call in mock_debug.call_args_list for arg in call.args] + + assert any("[HTTP-REQ]" in msg for msg in all_logged_args) + assert any("[HTTP-RES-ERR]" in msg for msg in all_logged_args) + assert any("[HTTP-RES-ERR-BODY] %s" in msg for msg in all_logged_args) + + # 验证实际的响应体被当做变量传入 + assert any("502 Bad Gateway" in msg for msg in all_logged_args) + + +@responses.activate() +def test_session_debug_logging_truncation(session, monkeypatch): + """测试开启 DEBUG 时,超长请求/响应体会按预期截断,防止刷屏""" + monkeypatch.setattr("swanlab.sdk.internal.core_python.client.session.helper.env.DEBUG", True) + mock_debug = MagicMock() + monkeypatch.setattr("swanlab.sdk.internal.core_python.client.session.log.debug", mock_debug) + + # 构造长度为 2000 的超长字符串 + long_string = "a" * 2000 + responses.add( + responses.POST, + "http://api.test/debug-truncate", + body=long_string, + status=200, + ) + + session.post("http://api.test/debug-truncate", data=long_string) + + truncation_marker = " ... (truncated)" + + # 直接在所有打出的变量里寻找包含 truncation_marker 的长字符串 + logged_vars = [str(arg) for call in mock_debug.call_args_list for arg in call.args] + truncated_bodies = [msg for msg in logged_vars if truncation_marker in msg] + + # 预期有两个被截断的 body (请求体 和 响应体) + assert len(truncated_bodies) == 2 + for body in truncated_bodies: + # 总长度应等于 1000 + 截断后缀长度 + assert len(body) == 1000 + len(truncation_marker) diff --git a/tests/unit/sdk/internal/pkg/test_log.py b/tests/unit/sdk/internal/pkg/test_log.py new file mode 100644 index 000000000..d82c57993 --- /dev/null +++ b/tests/unit/sdk/internal/pkg/test_log.py @@ -0,0 +1,103 @@ +""" +@author: cunyue +@file: test_log.py +@time: 2026/3/10 +@description: 测试 swanlab.sdk.pkg.log 诊断日志模块 (极简黑盒版) +""" + +import os +import stat + +import pytest + +import swanlab.sdk.internal.pkg.log as log_mod + + +def test_log_lifecycle_and_output(tmp_path): + """测试核心业务流:缓冲阶段记录 -> 绑定文件触发落盘 -> 直接写入文件""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + + # 1. 缓冲阶段(尚未 bindfile) + log_mod.debug("buffered debug") + log_mod.info("buffered info") + + # 2. 绑定文件(之前的缓冲应自动落盘) + log_mod.bindfile(log_dir) + + # 3. 绑定后写入(应直接落盘) + log_mod.warning("direct warning") + log_mod.error("direct error") + log_mod.critical("direct critical") + + # 验证最终落盘内容 + log_file = log_dir / "debug.log" + content = log_file.read_text(encoding="utf-8") + + assert "buffered debug" in content + assert "buffered info" in content + assert "Diagnostic log bound to file" in content + assert "direct warning" in content + assert "direct error" in content + assert "direct critical" in content + + +def test_bindfile_invalid_dir(tmp_path): + """测试绑定到不存在的目录时抛出 FileNotFoundError""" + with pytest.raises(FileNotFoundError, match="Log directory does not exist"): + log_mod.bindfile(tmp_path / "not_exist") + + +def test_bindfile_is_idempotent(tmp_path): + """测试重复调用 bindfile 的幂等性(不报错且不重复创建资源)""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + + # 第一次绑定 + log_mod.bindfile(log_dir) + log_mod.info("first bound msg") + + # 第二次重复绑定(应静默忽略) + log_mod.bindfile(log_dir) + log_mod.info("second bound msg") + + content = (log_dir / "debug.log").read_text(encoding="utf-8") + + assert "first bound msg" in content + assert "second bound msg" in content + + +def test_log_format(tmp_path): + """测试暴露的日志格式符合预期""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + log_mod.bindfile(log_dir) + + log_mod.error("check format") + + content = (log_dir / "debug.log").read_text(encoding="utf-8") + # 预期格式:时间 | ERROR | swanlab.internal | 消息 + assert " | ERROR | swanlab.internal | check format" in content + + +def test_secure_file_permissions(tmp_path): + """测试生成的诊断日志文件是否具备安全的 0600 权限 (仅限 POSIX 系统)""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + + # 绑定并写入一条日志,确保底层 _open 方法被真实触发 + log_mod.bindfile(log_dir) + log_mod.info("trigger file creation for permission check") + + log_file = log_dir / "debug.log" + assert log_file.exists() + + # 仅在类 Unix 系统下校验 POSIX 权限 + if os.name == "posix": + # 获取文件的完整状态信息 + mode = os.stat(log_file).st_mode + # 提取底部的八进制权限位 + permissions = stat.S_IMODE(mode) + + # 验证权限是否被严格限制为 0o600 (即 rw-------,仅所有者可读写) + assert permissions == 0o600