wehub-resource-sync
650 lines
26 KiB
Python
650 lines
26 KiB
Python
"""
|
||
日志系统特征测试（characterization tests）
|
||
|
||
固化 myboot/core/logger.py 的「当前实际行为」，作为后续重构的兼容性闸门。
|
||
失败时优先修改测试以匹配源码行为（除非源码行为本身是回归）。
|
||
|
||
注意:
|
||
- loguru 是全局单例，每个测试添加的 sink 必须移除（见 _loguru_isolation fixture）。
|
||
- setup_logging / setup_worker_logging 会调用 loguru_logger.remove() 清掉所有
|
||
handler（包括测试 sink），因此捕获 sink 必须在 setup 调用之后添加，
|
||
或者通过 capsys 捕获 stdout。
|
||
- 始终显式传入 Dynaconf 配置对象，避免 get_settings() 全局单例读取
|
||
项目根目录的 conf/config.yaml 造成测试间/环境间耦合。
|
||
"""
|
||
|
||
import inspect
|
||
import json
|
||
import logging
|
||
|
||
import pytest
|
||
from dynaconf import Dynaconf
|
||
from loguru import logger as loguru_logger
|
||
|
||
from myboot.core.logger import (
|
||
Logger,
|
||
_configure_third_party_loggers,
|
||
_convert_logging_format_to_loguru,
|
||
_get_worker_info,
|
||
_parse_json_config,
|
||
configure_worker_logger,
|
||
get_logger,
|
||
setup_logging,
|
||
setup_worker_logging,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _loguru_isolation():
|
||
"""保证测试顺序无关性。
|
||
|
||
loguru 是全局单例；setup_logging/setup_worker_logging 会 remove() 所有
|
||
handler 并 configure() 全局 extra。每个测试结束后清掉所有 handler 并
|
||
重置 extra，使下一个测试从确定状态开始（无 handler、空 extra）。
|
||
"""
|
||
yield
|
||
loguru_logger.remove()
|
||
loguru_logger.configure(extra={})
|
||
|
||
|
||
@pytest.fixture
|
||
def capture_records():
|
||
"""添加内存 sink 捕获 loguru record，返回 (records, add) 工具。
|
||
|
||
add() 在调用时刻挂 sink（便于在 setup_* 调用之后再挂，避免被
|
||
loguru_logger.remove() 清掉），finally 由 _loguru_isolation 兜底移除。
|
||
"""
|
||
records = []
|
||
handler_ids = []
|
||
|
||
def _add(level="DEBUG"):
|
||
hid = loguru_logger.add(
|
||
lambda message: records.append(message),
|
||
format="{message}",
|
||
level=level,
|
||
)
|
||
handler_ids.append(hid)
|
||
return hid
|
||
|
||
try:
|
||
yield records, _add
|
||
finally:
|
||
for hid in handler_ids:
|
||
try:
|
||
loguru_logger.remove(hid)
|
||
except ValueError:
|
||
# 已被 setup_* 的 remove() 清掉
|
||
pass
|
||
|
||
|
||
def _make_config(tmp_path, yaml_content: str) -> Dynaconf:
|
||
"""从 YAML 字符串创建独立的 Dynaconf 配置对象（不经过全局 get_settings）"""
|
||
config_path = tmp_path / "logger_test_config.yaml"
|
||
config_path.write_text(yaml_content, encoding="utf-8")
|
||
return Dynaconf(settings_files=[str(config_path)])
|
||
|
||
|
||
def _messages(records):
|
||
"""提取捕获到的纯文本消息列表（去掉末尾换行）"""
|
||
return [str(m).rstrip("\n") for m in records]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 1. issue #5 回归：setup_worker_logging 应遵循配置的日志级别（已修复）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSetupWorkerLoggingLevel:
|
||
def test_signature_accepts_config(self):
|
||
"""修复后的签名：setup_worker_logging(worker_id, total_workers, config=None)"""
|
||
params = inspect.signature(setup_worker_logging).parameters
|
||
assert list(params) == ["worker_id", "total_workers", "config"]
|
||
assert params["config"].default is None
|
||
|
||
def test_respects_configured_warning_level(self, tmp_path, capsys):
|
||
"""配置 WARNING 级别时，DEBUG/INFO 被抑制（不再硬编码 DEBUG）"""
|
||
cfg = _make_config(tmp_path, "logging:\n level: WARNING\n")
|
||
setup_worker_logging(worker_id=1, total_workers=2, config=cfg)
|
||
|
||
loguru_logger.debug("debug-should-not-appear-xyz")
|
||
loguru_logger.info("info-should-not-appear-xyz")
|
||
loguru_logger.warning("warning-should-appear-xyz")
|
||
|
||
out = capsys.readouterr().out
|
||
assert "debug-should-not-appear-xyz" not in out
|
||
assert "info-should-not-appear-xyz" not in out
|
||
assert "warning-should-appear-xyz" in out
|
||
|
||
def test_respects_configured_debug_level(self, tmp_path, capsys):
|
||
"""配置 DEBUG 级别时，DEBUG 消息正常输出"""
|
||
cfg = _make_config(tmp_path, "logging:\n level: DEBUG\n")
|
||
setup_worker_logging(worker_id=1, total_workers=2, config=cfg)
|
||
|
||
loguru_logger.debug("debug-should-appear-xyz")
|
||
out = capsys.readouterr().out
|
||
assert "debug-should-appear-xyz" in out
|
||
|
||
def test_default_level_is_info_when_config_missing_key(self, tmp_path, capsys):
|
||
"""配置中无 logging.level 时，当前默认级别为 INFO"""
|
||
cfg = _make_config(tmp_path, "app:\n name: demo\n")
|
||
setup_worker_logging(worker_id=1, total_workers=2, config=cfg)
|
||
|
||
loguru_logger.debug("debug-hidden-by-default-xyz")
|
||
loguru_logger.info("info-shown-by-default-xyz")
|
||
out = capsys.readouterr().out
|
||
assert "debug-hidden-by-default-xyz" not in out
|
||
assert "info-shown-by-default-xyz" in out
|
||
|
||
def test_output_contains_worker_tag(self, tmp_path, capsys):
|
||
"""worker 日志格式包含 Worker-{id}/{total} 标识"""
|
||
cfg = _make_config(tmp_path, "logging:\n level: INFO\n")
|
||
setup_worker_logging(worker_id=3, total_workers=8, config=cfg)
|
||
|
||
loguru_logger.info("hello-from-worker-xyz")
|
||
out = capsys.readouterr().out
|
||
assert "Worker-3/8" in out
|
||
assert "hello-from-worker-xyz" in out
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 2. Logger 兼容类
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestLoggerCompatClass:
|
||
def test_instantiation_stores_name(self):
|
||
log = Logger("svc")
|
||
assert log.name == "svc"
|
||
|
||
def test_default_name_is_app(self):
|
||
assert Logger().name == "app"
|
||
|
||
def test_all_level_methods_callable_without_error(self, capture_records):
|
||
records, add = capture_records
|
||
add(level="DEBUG")
|
||
log = Logger("compat")
|
||
log.debug("d-msg")
|
||
log.info("i-msg")
|
||
log.warning("w-msg")
|
||
log.error("e-msg")
|
||
log.critical("c-msg")
|
||
# exception() 在无活动异常时也不抛错（loguru 行为）
|
||
log.exception("x-msg")
|
||
msgs = _messages(records)
|
||
assert msgs[:5] == ["d-msg", "i-msg", "w-msg", "e-msg", "c-msg"]
|
||
# 可疑现状：无活动异常时 exception() 仍输出异常段，
|
||
# 格式化文本末尾附加 "NoneType: None"
|
||
assert msgs[5] == "x-msg\nNoneType: None"
|
||
|
||
def test_bind_name_in_extra(self):
|
||
"""Logger 内部 _logger 是 bind(name=...) 后的 loguru logger，
|
||
name 进入 record["extra"]["name"]（不影响 record["name"] 即模块名）"""
|
||
records = []
|
||
hid = loguru_logger.add(
|
||
lambda m: records.append(m.record), format="{message}", level="DEBUG"
|
||
)
|
||
try:
|
||
Logger("my-component").info("bound message")
|
||
finally:
|
||
loguru_logger.remove(hid)
|
||
|
||
assert len(records) == 1
|
||
assert records[0]["extra"]["name"] == "my-component"
|
||
# 可疑现状：record["name"]（loguru 内置，模块路径）不受 bind 影响，
|
||
# 仍为调用方模块名，而非传入的 name —— 文本格式 {name} 显示的是模块名
|
||
assert records[0]["name"] != "my-component"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 3. get_logger
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGetLogger:
|
||
def test_returns_logger_bound_with_name(self):
|
||
records = []
|
||
hid = loguru_logger.add(
|
||
lambda m: records.append(m.record), format="{message}", level="DEBUG"
|
||
)
|
||
try:
|
||
log = get_logger("api")
|
||
log.info("from get_logger")
|
||
finally:
|
||
loguru_logger.remove(hid)
|
||
|
||
assert len(records) == 1
|
||
assert records[0]["extra"]["name"] == "api"
|
||
|
||
def test_default_name_is_app(self):
|
||
records = []
|
||
hid = loguru_logger.add(
|
||
lambda m: records.append(m.record), format="{message}", level="DEBUG"
|
||
)
|
||
try:
|
||
get_logger().info("default name")
|
||
finally:
|
||
loguru_logger.remove(hid)
|
||
|
||
assert records[0]["extra"]["name"] == "app"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 4. loguru {} 占位符格式化
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestBracePlaceholderFormatting:
|
||
def test_positional_brace_formatting(self, capture_records):
|
||
records, add = capture_records
|
||
add()
|
||
loguru_logger.info("hello {}", "world")
|
||
assert _messages(records) == ["hello world"]
|
||
|
||
def test_named_brace_formatting_via_kwargs(self, capture_records):
|
||
records, add = capture_records
|
||
add()
|
||
loguru_logger.info("hello {who}", who="myboot")
|
||
assert _messages(records) == ["hello myboot"]
|
||
|
||
def test_logger_compat_class_supports_brace_args(self, capture_records):
|
||
"""Logger 兼容类透传 *args，因此 {} 占位符同样可用"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("count={}", 42)
|
||
assert _messages(records) == ["count=42"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 5. %s 风格调用（issue #12：Logger 兼容类同时支持 % 与 {} 占位符）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPercentStyleSupport:
|
||
def test_percent_s_is_formatted(self, capture_records):
|
||
"""issue #12：原行为是 %s 字面量输出、参数静默丢弃；
|
||
修复后 Logger 兼容类对 % 占位符做预格式化"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("hello %s", "world")
|
||
assert _messages(records) == ["hello world"]
|
||
|
||
def test_percent_multiple_args_and_types(self, capture_records):
|
||
"""issue #12：%s/%d 多参数预格式化"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("user=%s count=%d", "alice", 3)
|
||
assert _messages(records) == ["user=alice count=3"]
|
||
|
||
def test_percent_works_on_all_level_methods(self, capture_records):
|
||
"""issue #12：六个级别方法统一支持 % 占位符"""
|
||
records, add = capture_records
|
||
add(level="DEBUG")
|
||
log = Logger("x")
|
||
log.debug("d=%s", 1)
|
||
log.info("i=%s", 2)
|
||
log.warning("w=%s", 3)
|
||
log.error("e=%s", 4)
|
||
log.critical("c=%s", 5)
|
||
log.exception("x=%s", 6)
|
||
msgs = _messages(records)
|
||
assert msgs[:5] == ["d=1", "i=2", "w=3", "e=4", "c=5"]
|
||
# exception() 无活动异常时附加 "NoneType: None"（既有 loguru 行为）
|
||
assert msgs[5] == "x=6\nNoneType: None"
|
||
|
||
def test_double_percent_escape(self, capture_records):
|
||
"""issue #12：%% 是转义不是占位符，% 预格式化后输出单个 %"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("progress 100%% by %s", "worker")
|
||
assert _messages(records) == ["progress 100% by worker"]
|
||
|
||
def test_double_percent_only_without_real_placeholder(self, capture_records):
|
||
"""issue #12：消息只含 %%（无真正 % 占位符）且带参数时，
|
||
不做 % 预格式化，参数交给 loguru（无 {} 则被忽略），%% 原样保留"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("progress 100%%", "ignored")
|
||
assert _messages(records) == ["progress 100%%"]
|
||
|
||
def test_mixed_percent_and_brace_prefers_brace(self, capture_records):
|
||
"""issue #12 设计决策：消息同时含 % 和 {} 占位符时，{} 优先，
|
||
原样交给 loguru 的 str.format（保持 loguru 原生语义），% 保留字面量"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("pct=%s val={}", 42)
|
||
assert _messages(records) == ["pct=%s val=42"]
|
||
|
||
def test_percent_args_mismatch_falls_back_without_raising(self, capture_records):
|
||
"""issue #12：% 预格式化参数不匹配时回退为原样输出，不向调用方抛错"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("a=%s b=%s", "only-one")
|
||
assert _messages(records) == ["a=%s b=%s"]
|
||
|
||
def test_lone_brace_with_args_does_not_raise(self, capture_records):
|
||
"""issue #12：原行为是孤立 { 带参数时 loguru 的 str.format 异常
|
||
直接传播；修复后走 % 预格式化路径（无完整 {} 占位符），不抛错"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("ratio %s {bad", "x")
|
||
assert _messages(records) == ["ratio x {bad"]
|
||
|
||
def test_lone_brace_without_percent_degrades_safely(self, capture_records):
|
||
"""issue #12：孤立 { 且无 % 占位符但带参数时，loguru 内部
|
||
str.format 抛错被捕获，降级为无参输出消息本体，不抛错"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("oops {bad", "x")
|
||
assert _messages(records) == ["oops {bad"]
|
||
|
||
def test_brace_format_failure_degrades_safely(self, capture_records):
|
||
"""issue #12：{} 路径下 str.format 仍可能失败（如孤立 { 与 {} 共存），
|
||
捕获后降级输出消息本体，不抛错"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("{bad and {}", "x")
|
||
assert _messages(records) == ["{bad and {}"]
|
||
|
||
def test_percent_s_on_raw_loguru_logger_unchanged(self, capture_records):
|
||
"""裸 loguru logger 不受 issue #12 影响：仍原样输出 %s，参数被丢弃"""
|
||
records, add = capture_records
|
||
add()
|
||
loguru_logger.info("value=%s", 123)
|
||
assert _messages(records) == ["value=%s"]
|
||
|
||
def test_raw_loguru_brace_chars_still_raise(self, capture_records):
|
||
"""裸 loguru logger 不受 issue #12 影响：含未转义 { } 且带参数时
|
||
str.format() 异常仍直接传播"""
|
||
records, add = capture_records
|
||
add()
|
||
with pytest.raises((ValueError, KeyError, IndexError)):
|
||
loguru_logger.info("ratio %s {bad", "x")
|
||
|
||
def test_message_without_args_not_formatted(self, capture_records):
|
||
"""无参数时不做任何格式化，含 % 或 {} 的消息原样输出（Logger 类与裸 loguru 一致）"""
|
||
records, add = capture_records
|
||
add()
|
||
loguru_logger.info("progress 100% {not-a-placeholder}")
|
||
Logger("x").info("progress 100% {not-a-placeholder}")
|
||
assert _messages(records) == [
|
||
"progress 100% {not-a-placeholder}",
|
||
"progress 100% {not-a-placeholder}",
|
||
]
|
||
|
||
|
||
class TestBracePathNoRegression:
|
||
"""issue #12 回归保护：{} 路径与命名 kwargs 用法保持不变"""
|
||
|
||
def test_logger_brace_positional_not_regressed(self, capture_records):
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("count={} name={}", 42, "svc")
|
||
assert _messages(records) == ["count=42 name=svc"]
|
||
|
||
def test_logger_named_kwargs_not_regressed(self, capture_records):
|
||
"""命名 {name} 占位符通过 kwargs 透传给 loguru"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("hello {who}", who="myboot")
|
||
assert _messages(records) == ["hello myboot"]
|
||
|
||
def test_logger_named_kwargs_with_positional_args(self, capture_records):
|
||
"""位置 {} 与命名 {name} 混用时 args/kwargs 均透传"""
|
||
records, add = capture_records
|
||
add()
|
||
Logger("x").info("{} meets {who}", "alice", who="bob")
|
||
assert _messages(records) == ["alice meets bob"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 6. JSON 日志格式
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestJsonLogging:
|
||
def test_setup_logging_json_outputs_valid_json(self, tmp_path, capsys, clean_myboot_env):
|
||
cfg = _make_config(tmp_path, "logging:\n level: INFO\n json: true\n")
|
||
setup_logging(config=cfg)
|
||
|
||
loguru_logger.info("json hello")
|
||
out = capsys.readouterr().out.strip()
|
||
|
||
parsed = json.loads(out)
|
||
assert parsed["record"]["message"] == "json hello"
|
||
assert parsed["record"]["level"]["name"] == "INFO"
|
||
# setup_logging 配置了全局 extra worker（单进程为 "Main"）
|
||
assert parsed["record"]["extra"]["worker"] == "Main"
|
||
# serialize=True 时同时包含人类可读的 text 字段
|
||
assert "json hello" in parsed["text"]
|
||
|
||
def test_json_config_accepts_string_true(self, tmp_path, capsys, clean_myboot_env):
|
||
"""logging.json 为字符串 "true" 时同样启用 JSON（_parse_json_config）"""
|
||
cfg = _make_config(tmp_path, 'logging:\n level: INFO\n json: "true"\n')
|
||
setup_logging(config=cfg)
|
||
|
||
loguru_logger.info("string-true json")
|
||
out = capsys.readouterr().out.strip()
|
||
assert json.loads(out)["record"]["message"] == "string-true json"
|
||
|
||
def test_parse_json_config_truth_table(self):
|
||
"""固化 _parse_json_config 的当前真值表"""
|
||
assert _parse_json_config(True) is True
|
||
assert _parse_json_config(False) is False
|
||
assert _parse_json_config("true") is True
|
||
assert _parse_json_config("TRUE") is True
|
||
assert _parse_json_config("1") is True
|
||
assert _parse_json_config("yes") is True
|
||
assert _parse_json_config("on") is True
|
||
assert _parse_json_config("false") is False
|
||
assert _parse_json_config("off") is False
|
||
# 可疑现状：任意其他字符串（如 "enabled"）一律为 False
|
||
assert _parse_json_config("enabled") is False
|
||
# 可疑现状：非 bool/str 走 bool() 转换，整数 1 为 True
|
||
assert _parse_json_config(1) is True
|
||
assert _parse_json_config(0) is False
|
||
assert _parse_json_config(None) is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 7. 第三方库日志级别控制
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestThirdPartyLoggerControl:
|
||
def test_configure_third_party_loggers_sets_std_logging_level(self):
|
||
name = "myboot_test_3rdparty_a"
|
||
try:
|
||
_configure_third_party_loggers({name: "WARNING"})
|
||
assert logging.getLogger(name).level == logging.WARNING
|
||
|
||
_configure_third_party_loggers({name: "error"}) # 大小写不敏感
|
||
assert logging.getLogger(name).level == logging.ERROR
|
||
finally:
|
||
logging.getLogger(name).setLevel(logging.NOTSET)
|
||
|
||
def test_unknown_level_falls_back_to_info(self):
|
||
"""可疑现状：无效级别名静默回退为 INFO（getattr 默认值），不报错"""
|
||
name = "myboot_test_3rdparty_b"
|
||
try:
|
||
_configure_third_party_loggers({name: "NOT_A_LEVEL"})
|
||
assert logging.getLogger(name).level == logging.INFO
|
||
finally:
|
||
logging.getLogger(name).setLevel(logging.NOTSET)
|
||
|
||
def test_non_dict_and_non_str_values_ignored(self):
|
||
name = "myboot_test_3rdparty_c"
|
||
try:
|
||
# 非 dict 输入：直接 return，不抛错
|
||
_configure_third_party_loggers("not-a-dict")
|
||
_configure_third_party_loggers(None)
|
||
# 可疑现状：级别值非字符串（如 int 30）被静默忽略
|
||
_configure_third_party_loggers({name: 30})
|
||
assert logging.getLogger(name).level == logging.NOTSET
|
||
finally:
|
||
logging.getLogger(name).setLevel(logging.NOTSET)
|
||
|
||
def test_setup_logging_applies_third_party_config(self, tmp_path, capsys, clean_myboot_env):
|
||
name = "myboot_test_3rdparty_d"
|
||
cfg = _make_config(
|
||
tmp_path,
|
||
f"logging:\n level: INFO\n third_party:\n {name}: ERROR\n",
|
||
)
|
||
try:
|
||
setup_logging(config=cfg)
|
||
assert logging.getLogger(name).level == logging.ERROR
|
||
finally:
|
||
logging.getLogger(name).setLevel(logging.NOTSET)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# setup_logging 整体行为
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSetupLogging:
|
||
def test_respects_configured_level(self, tmp_path, capsys, clean_myboot_env):
|
||
cfg = _make_config(tmp_path, "logging:\n level: ERROR\n")
|
||
setup_logging(config=cfg)
|
||
|
||
loguru_logger.info("info-suppressed-xyz")
|
||
loguru_logger.error("error-shown-xyz")
|
||
out = capsys.readouterr().out
|
||
assert "info-suppressed-xyz" not in out
|
||
assert "error-shown-xyz" in out
|
||
|
||
def test_removes_preexisting_handlers(self, tmp_path, capsys, clean_myboot_env):
|
||
"""setup_logging 调用 loguru_logger.remove() 清掉之前的所有 sink"""
|
||
before = []
|
||
loguru_logger.add(lambda m: before.append(m), format="{message}")
|
||
|
||
cfg = _make_config(tmp_path, "logging:\n level: INFO\n")
|
||
setup_logging(config=cfg)
|
||
|
||
loguru_logger.info("after-setup-xyz")
|
||
assert before == [] # 之前的 sink 已被移除，收不到任何消息
|
||
assert "after-setup-xyz" in capsys.readouterr().out
|
||
|
||
def test_single_worker_uses_simple_format_without_worker_tag(
|
||
self, tmp_path, capsys, clean_myboot_env
|
||
):
|
||
"""无 MYBOOT_WORKER_COUNT 环境变量时使用简单格式，不含 worker 标识"""
|
||
cfg = _make_config(tmp_path, "logging:\n level: INFO\n")
|
||
setup_logging(config=cfg)
|
||
|
||
loguru_logger.info("simple-format-xyz")
|
||
out = capsys.readouterr().out
|
||
assert "simple-format-xyz" in out
|
||
assert "Main" not in out
|
||
assert "Worker-" not in out
|
||
|
||
def test_multi_worker_env_enables_worker_tag(self, tmp_path, capsys, clean_myboot_env):
|
||
clean_myboot_env.setenv("MYBOOT_WORKER_COUNT", "4")
|
||
clean_myboot_env.setenv("MYBOOT_WORKER_ID", "2")
|
||
cfg = _make_config(tmp_path, "logging:\n level: INFO\n")
|
||
setup_logging(config=cfg)
|
||
|
||
loguru_logger.info("tagged-xyz")
|
||
out = capsys.readouterr().out
|
||
assert "Worker-2/4" in out
|
||
assert "tagged-xyz" in out
|
||
|
||
def test_enable_worker_info_false_suppresses_worker_tag(
|
||
self, tmp_path, capsys, clean_myboot_env
|
||
):
|
||
clean_myboot_env.setenv("MYBOOT_WORKER_COUNT", "4")
|
||
clean_myboot_env.setenv("MYBOOT_WORKER_ID", "2")
|
||
cfg = _make_config(tmp_path, "logging:\n level: INFO\n")
|
||
setup_logging(config=cfg, enable_worker_info=False)
|
||
|
||
loguru_logger.info("untagged-xyz")
|
||
out = capsys.readouterr().out
|
||
assert "Worker-2/4" not in out
|
||
assert "untagged-xyz" in out
|
||
|
||
def test_custom_logging_format_converted_to_loguru(self, tmp_path, capsys, clean_myboot_env):
|
||
"""logging.format 支持标准 logging 风格 %(...)s 占位符并被转换"""
|
||
cfg = _make_config(
|
||
tmp_path,
|
||
'logging:\n level: INFO\n format: "%(levelname)s | %(message)s"\n',
|
||
)
|
||
setup_logging(config=cfg)
|
||
|
||
loguru_logger.info("converted-format-xyz")
|
||
out = capsys.readouterr().out
|
||
assert "INFO" in out
|
||
assert "converted-format-xyz" in out
|
||
|
||
def test_file_handler_writes_to_configured_path(self, tmp_path, capsys, clean_myboot_env):
|
||
log_file = tmp_path / "logs" / "app.log"
|
||
cfg = _make_config(
|
||
tmp_path,
|
||
f"logging:\n level: INFO\n file: {json.dumps(str(log_file))}\n",
|
||
)
|
||
setup_logging(config=cfg)
|
||
|
||
loguru_logger.info("written-to-file-xyz")
|
||
# 移除 handler 以确保文件被 flush/关闭后再读取（Windows）
|
||
loguru_logger.remove()
|
||
|
||
assert log_file.exists() # 父目录由 _add_file_handler 自动创建
|
||
content = log_file.read_text(encoding="utf-8")
|
||
assert "written-to-file-xyz" in content
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# configure_worker_logger 与辅助函数
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestWorkerContextHelpers:
|
||
def test_configure_worker_logger_sets_extra_worker(self):
|
||
configure_worker_logger(worker_id=3, total_workers=8)
|
||
|
||
records = []
|
||
hid = loguru_logger.add(
|
||
lambda m: records.append(m.record), format="{message}", level="DEBUG"
|
||
)
|
||
try:
|
||
loguru_logger.info("ctx check")
|
||
finally:
|
||
loguru_logger.remove(hid)
|
||
|
||
assert records[0]["extra"]["worker"] == "Worker-3/8"
|
||
|
||
def test_get_worker_info_main_without_env(self, clean_myboot_env):
|
||
assert _get_worker_info() == "Main"
|
||
|
||
def test_get_worker_info_with_env(self, clean_myboot_env):
|
||
clean_myboot_env.setenv("MYBOOT_WORKER_ID", "2")
|
||
clean_myboot_env.setenv("MYBOOT_WORKER_COUNT", "4")
|
||
assert _get_worker_info() == "Worker-2/4"
|
||
|
||
def test_get_worker_info_requires_both_env_vars(self, clean_myboot_env):
|
||
"""可疑现状：只设置 WORKER_ID 而无 WORKER_COUNT 时返回 "Main" """
|
||
clean_myboot_env.setenv("MYBOOT_WORKER_ID", "2")
|
||
assert _get_worker_info() == "Main"
|
||
|
||
|
||
class TestFormatConversion:
|
||
def test_standard_logging_placeholders_converted(self):
|
||
converted = _convert_logging_format_to_loguru(
|
||
"%(asctime)s %(name)s %(levelname)s %(message)s "
|
||
"%(filename)s %(funcName)s %(lineno)d"
|
||
)
|
||
assert converted == (
|
||
"{time:YYYY-MM-DD HH:mm:ss} {name} {level: <8} {message} "
|
||
"{file.name} {function} {line}"
|
||
)
|
||
|
||
def test_unknown_placeholders_left_untouched(self):
|
||
"""可疑现状：映射表之外的占位符（如 %(thread)d）原样保留，
|
||
进入 loguru format 后会原样输出在每条日志中"""
|
||
assert _convert_logging_format_to_loguru("%(thread)d %(message)s") == (
|
||
"%(thread)d {message}"
|
||
)
|