FinDataHub
11d36d1c1e
问题 - 索引中该目录记为小写 `scripts/`，工作区实为 `Scripts/`。因 core.ignorecase=true， 本地 `git diff -- Scripts/bin/x.py` 会**静默返回空**（看似无改动）， 且在大小写敏感环境（Linux CI / Docker 挂载）会检出 `scripts/`， 与全仓所有 `Scripts/...` 引用不匹配。 - 该目录此前 10 个脚本（check_python_version.py、perf_snapshot.py、 install_collector_launchd.py、mysql_toggle.sh 等）从未入库， 而 CI 第 31 行正调用 `Scripts/bin/check_python_version.py --strict` —— 全新检出必然失败。 改动 - 索引路径统一为 `Scripts/`（工作区文件未移动，仅索引改名）。 - 补齐上述 10 个未跟踪脚本及其测试。 影响 - 修正前索引 35 条（全为小写路径），修正后 45 条。
550 lines
19 KiB
Python
550 lines
19 KiB
Python
# coding=utf-8
|
||
"""common.py 公共配置与工具模块单元测试.
|
||
|
||
重点覆盖：
|
||
- database_url：SQLAlchemy URL 构造 + 密码 URL 编码
|
||
- load_env：.env 文件解析（注释、引号、空行、已有变量不覆盖）
|
||
- PidManager：PID 文件写入/状态检查/停止
|
||
- log_*：分级日志输出控制
|
||
- is_debug_enabled：debug 级别判断
|
||
- _resolve_bundled_paths：路径解析回退逻辑
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
import pytest
|
||
|
||
# 将 Scripts/bin 加入 sys.path
|
||
_bin_dir = os.path.join(os.path.dirname(__file__), "..")
|
||
if _bin_dir not in sys.path:
|
||
sys.path.insert(0, _bin_dir)
|
||
|
||
|
||
# ── database_url ─────────────────────────────────────────
|
||
|
||
|
||
class TestDatabaseUrl:
|
||
"""SQLAlchemy DATABASE_URL 构造测试."""
|
||
|
||
def test_basic_url(self):
|
||
"""基本 URL 格式正确."""
|
||
from common import database_url
|
||
|
||
cfg = {
|
||
"host": "localhost",
|
||
"port": 3306,
|
||
"user": "root",
|
||
"password": "pass",
|
||
"database": "test_db",
|
||
}
|
||
url = database_url(cfg)
|
||
assert url == "mysql+pymysql://root:pass@localhost:3306/test_db?charset=utf8mb4"
|
||
|
||
def test_password_with_special_chars_encoded(self):
|
||
"""含特殊字符的密码被 URL 编码."""
|
||
from common import database_url
|
||
|
||
cfg = {
|
||
"host": "localhost",
|
||
"port": 3306,
|
||
"user": "root",
|
||
"password": "p@ss/w0rd!",
|
||
"database": "test_db",
|
||
}
|
||
url = database_url(cfg)
|
||
# @ → %40, / → %2F, ! → %21
|
||
assert "p%40ss%2Fw0rd%21" in url
|
||
assert "p@ss" not in url # 原始密码不应出现在 URL 中
|
||
|
||
def test_empty_password(self):
|
||
"""空密码的 URL 不含密码段."""
|
||
from common import database_url
|
||
|
||
cfg = {
|
||
"host": "localhost",
|
||
"port": 3306,
|
||
"user": "root",
|
||
"password": "",
|
||
"database": "test_db",
|
||
}
|
||
url = database_url(cfg)
|
||
assert url == "mysql+pymysql://root:@localhost:3306/test_db?charset=utf8mb4"
|
||
|
||
def test_default_cfg_when_none(self):
|
||
"""cfg=None 时使用默认 MYSQL_CFG."""
|
||
from common import database_url, MYSQL_CFG
|
||
|
||
url = database_url()
|
||
assert url.startswith("mysql+pymysql://")
|
||
assert MYSQL_CFG["database"] in url
|
||
|
||
def test_custom_port(self):
|
||
"""自定义端口出现在 URL 中."""
|
||
from common import database_url
|
||
|
||
cfg = {
|
||
"host": "db.example.com",
|
||
"port": 3307,
|
||
"user": "admin",
|
||
"password": "secret",
|
||
"database": "prod",
|
||
}
|
||
url = database_url(cfg)
|
||
assert ":3307/" in url
|
||
|
||
|
||
# ── load_env ─────────────────────────────────────────────
|
||
|
||
|
||
class TestLoadEnv:
|
||
""" .env 文件加载测试."""
|
||
|
||
def test_basic_env_loading(self, tmp_path, monkeypatch):
|
||
"""基本 key=value 被加载到 os.environ."""
|
||
from common import load_env
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text("FOO=bar\nBAZ=qux\n")
|
||
monkeypatch.delenv("FOO", raising=False)
|
||
monkeypatch.delenv("BAZ", raising=False)
|
||
load_env(str(env_file))
|
||
assert os.environ.get("FOO") == "bar"
|
||
assert os.environ.get("BAZ") == "qux"
|
||
|
||
def test_comment_lines_ignored(self, tmp_path, monkeypatch):
|
||
"""以 # 开头的注释行被忽略."""
|
||
from common import load_env
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text("# This is a comment\nKEY=value\n")
|
||
monkeypatch.delenv("KEY", raising=False)
|
||
load_env(str(env_file))
|
||
assert os.environ.get("KEY") == "value"
|
||
|
||
def test_empty_lines_ignored(self, tmp_path, monkeypatch):
|
||
"""空行被忽略."""
|
||
from common import load_env
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text("\n\nKEY=value\n\n")
|
||
monkeypatch.delenv("KEY", raising=False)
|
||
load_env(str(env_file))
|
||
assert os.environ.get("KEY") == "value"
|
||
|
||
def test_no_equals_sign_ignored(self, tmp_path):
|
||
"""没有 = 号的行被忽略."""
|
||
from common import load_env
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text("NOT_A_CONFIG_LINE\nKEY=value\n")
|
||
load_env(str(env_file))
|
||
# 不应崩溃，KEY 应被加载
|
||
assert os.environ.get("KEY") == "value"
|
||
|
||
def test_quoted_values_stripped(self, tmp_path, monkeypatch):
|
||
"""双引号和单引号包裹的值被去除引号."""
|
||
from common import load_env
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text('DOUBLE="quoted_value"\nSINGLE=\'single_value\'\n')
|
||
monkeypatch.delenv("DOUBLE", raising=False)
|
||
monkeypatch.delenv("SINGLE", raising=False)
|
||
load_env(str(env_file))
|
||
assert os.environ.get("DOUBLE") == "quoted_value"
|
||
assert os.environ.get("SINGLE") == "single_value"
|
||
|
||
def test_existing_env_not_overwritten(self, tmp_path, monkeypatch):
|
||
"""已有环境变量不被覆盖."""
|
||
from common import load_env
|
||
|
||
monkeypatch.setenv("EXISTING_VAR", "original")
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text("EXISTING_VAR=new_value\n")
|
||
load_env(str(env_file))
|
||
assert os.environ.get("EXISTING_VAR") == "original"
|
||
|
||
def test_nonexistent_file_silent(self, tmp_path):
|
||
"""文件不存在时静默返回（不抛异常）."""
|
||
from common import load_env
|
||
|
||
load_env(str(tmp_path / "nonexistent.env")) # 不应抛异常
|
||
|
||
def test_value_with_equals_sign(self, tmp_path, monkeypatch):
|
||
"""值中包含 = 号时正确解析（split("=", 1)）."""
|
||
from common import load_env
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text("CONNECTION_STRING=host=db;user=admin;pass=123\n")
|
||
monkeypatch.delenv("CONNECTION_STRING", raising=False)
|
||
load_env(str(env_file))
|
||
assert os.environ.get("CONNECTION_STRING") == "host=db;user=admin;pass=123"
|
||
|
||
|
||
# ── PidManager ───────────────────────────────────────────
|
||
|
||
|
||
class TestPidManager:
|
||
"""PID 文件管理测试."""
|
||
|
||
def test_write_creates_file(self, tmp_path):
|
||
"""write() 创建 PID 文件."""
|
||
from common import PidManager
|
||
|
||
pid_file = str(tmp_path / "test.pid")
|
||
mgr = PidManager(pid_file)
|
||
mgr.write()
|
||
assert os.path.exists(pid_file)
|
||
content = Path(pid_file).read_text()
|
||
assert ":" in content # 格式: pid:timestamp
|
||
assert str(os.getpid()) in content
|
||
|
||
def test_status_no_file(self, tmp_path):
|
||
"""PID 文件不存在时返回 running=False."""
|
||
from common import PidManager
|
||
|
||
mgr = PidManager(str(tmp_path / "nonexistent.pid"))
|
||
status = mgr.status()
|
||
assert status["running"] is False
|
||
assert status["pid"] is None
|
||
assert status["started_at"] is None
|
||
|
||
def test_status_running_process(self, tmp_path):
|
||
"""当前进程 PID 标记为运行中."""
|
||
from common import PidManager
|
||
|
||
pid_file = str(tmp_path / "test.pid")
|
||
mgr = PidManager(pid_file)
|
||
mgr.write()
|
||
status = mgr.status()
|
||
assert status["running"] is True
|
||
assert status["pid"] == os.getpid()
|
||
assert status["started_at"] is not None
|
||
|
||
def test_status_preserves_full_timestamp(self, tmp_path):
|
||
"""status() 必须保留完整 ISO 时间戳，不得被截断到整点.
|
||
|
||
回归用例：write() 写 "pid:2026-09-17T18:12:13.463481"，旧实现用
|
||
split(":") 取 parts[1] 只得 "2026-09-17T18"，fromisoformat 解析成 18:00:00，
|
||
使下游 (now - started) > 600 在每小时第 10 分钟后恒真，stuck 被误判为卡死。
|
||
"""
|
||
from datetime import datetime
|
||
|
||
from common import PidManager
|
||
|
||
pid_file = str(tmp_path / "test.pid")
|
||
mgr = PidManager(pid_file)
|
||
mgr.write()
|
||
|
||
raw_ts = Path(pid_file).read_text().split(":", 1)[1]
|
||
status = mgr.status()
|
||
|
||
# 与文件内原始时间戳逐字一致（未被截断）
|
||
assert status["started_at"] == raw_ts
|
||
# 截断后的表现是长度退化为 "YYYY-MM-DDTHH"（13 字符）
|
||
assert len(status["started_at"]) > len("2024-01-01T00")
|
||
# 且能解析回真实时刻（分钟/秒不丢）
|
||
parsed = datetime.fromisoformat(status["started_at"])
|
||
assert abs((datetime.now() - parsed).total_seconds()) < 60
|
||
|
||
def test_status_dead_process_cleans_file(self, tmp_path):
|
||
"""死进程的 PID 文件被自动清理."""
|
||
from common import PidManager
|
||
|
||
pid_file = str(tmp_path / "test.pid")
|
||
# 写入一个不存在的 PID
|
||
Path(pid_file).write_text("99999999:2024-01-01T00:00:00")
|
||
mgr = PidManager(pid_file)
|
||
status = mgr.status()
|
||
assert status["running"] is False
|
||
# 文件应被删除
|
||
assert not os.path.exists(pid_file)
|
||
|
||
def test_status_invalid_pid_format(self, tmp_path):
|
||
"""PID 文件内容非数字时 pid=None."""
|
||
from common import PidManager
|
||
|
||
pid_file = str(tmp_path / "test.pid")
|
||
Path(pid_file).write_text("not_a_number:some_time")
|
||
mgr = PidManager(pid_file)
|
||
status = mgr.status()
|
||
assert status["running"] is False
|
||
assert status["pid"] is None
|
||
|
||
def test_stop_removes_file(self, tmp_path, monkeypatch):
|
||
"""stop() 删除 PID 文件."""
|
||
from common import PidManager
|
||
|
||
# write() 写入当前进程 PID，stop() 会 kill 该 PID → mock kill 防自杀
|
||
monkeypatch.setattr("os.kill", lambda *a, **k: None)
|
||
pid_file = str(tmp_path / "test.pid")
|
||
mgr = PidManager(pid_file)
|
||
mgr.write()
|
||
result = mgr.stop()
|
||
assert result is True
|
||
assert not os.path.exists(pid_file)
|
||
|
||
def test_stop_no_file_returns_false(self, tmp_path):
|
||
"""PID 文件不存在时 stop() 返回 False."""
|
||
from common import PidManager
|
||
|
||
mgr = PidManager(str(tmp_path / "nonexistent.pid"))
|
||
assert mgr.stop() is False
|
||
|
||
|
||
# ── 日志工具 ─────────────────────────────────────────────
|
||
|
||
|
||
class TestDebugLogging:
|
||
"""分级日志输出控制测试."""
|
||
|
||
def test_is_debug_enabled_when_debug(self, monkeypatch):
|
||
"""FDH_DEBUG=debug 时 is_debug_enabled() 返回 True."""
|
||
import importlib
|
||
|
||
monkeypatch.setenv("FDH_DEBUG", "debug")
|
||
import common
|
||
importlib.reload(common)
|
||
assert common.is_debug_enabled() is True
|
||
|
||
def test_is_debug_enabled_when_info(self, monkeypatch):
|
||
"""FDH_DEBUG=info 时 is_debug_enabled() 返回 False."""
|
||
import importlib
|
||
|
||
monkeypatch.setenv("FDH_DEBUG", "info")
|
||
import common
|
||
importlib.reload(common)
|
||
assert common.is_debug_enabled() is False
|
||
|
||
def test_is_debug_enabled_when_off(self, monkeypatch):
|
||
"""FDH_DEBUG=off 时 is_debug_enabled() 返回 False."""
|
||
import importlib
|
||
|
||
monkeypatch.setenv("FDH_DEBUG", "off")
|
||
import common
|
||
importlib.reload(common)
|
||
assert common.is_debug_enabled() is False
|
||
|
||
def test_log_debug_outputs_to_stderr(self, monkeypatch, capsys):
|
||
"""log_debug 在 debug 模式下输出到 stderr."""
|
||
import importlib
|
||
|
||
monkeypatch.setenv("FDH_DEBUG", "debug")
|
||
import common
|
||
importlib.reload(common)
|
||
common.log_debug("TestCategory", "test debug message")
|
||
captured = capsys.readouterr()
|
||
assert "test debug message" in captured.err
|
||
assert "DEBUG" in captured.err
|
||
|
||
def test_log_debug_silent_when_info(self, monkeypatch, capsys):
|
||
"""log_debug 在 info 模式下不输出."""
|
||
import importlib
|
||
|
||
monkeypatch.setenv("FDH_DEBUG", "info")
|
||
import common
|
||
importlib.reload(common)
|
||
common.log_debug("TestCategory", "should not appear")
|
||
captured = capsys.readouterr()
|
||
assert "should not appear" not in captured.err
|
||
|
||
def test_log_error_always_outputs(self, monkeypatch, capsys):
|
||
"""log_error 在所有非 off 模式下都输出."""
|
||
import importlib
|
||
|
||
monkeypatch.setenv("FDH_DEBUG", "info")
|
||
import common
|
||
importlib.reload(common)
|
||
common.log_error("TestCategory", "error message")
|
||
captured = capsys.readouterr()
|
||
assert "error message" in captured.err
|
||
assert "ERROR" in captured.err
|
||
|
||
def test_log_off_silences_all(self, monkeypatch, capsys):
|
||
"""FDH_DEBUG=off 时所有日志静默."""
|
||
import importlib
|
||
|
||
monkeypatch.setenv("FDH_DEBUG", "off")
|
||
import common
|
||
importlib.reload(common)
|
||
common.log_error("TestCategory", "should not appear")
|
||
common.log_info("TestCategory", "should not appear")
|
||
common.log_debug("TestCategory", "should not appear")
|
||
common.log_warn("TestCategory", "should not appear")
|
||
captured = capsys.readouterr()
|
||
assert captured.err == ""
|
||
|
||
|
||
# ── _resolve_bundled_paths ─────────────────────────────────
|
||
|
||
|
||
class TestResolveBundledPaths:
|
||
"""_resolve_bundled_paths 捆绑路径解析测试.
|
||
|
||
_resolve_bundled_paths 使用 resources_dir = Path(BIN_DIR).parent，
|
||
然后查找 resources_dir/kuaixun-service。因此测试需要在 BIN_DIR 的
|
||
父目录下创建 kuaixun-service 目录。
|
||
|
||
覆盖：
|
||
- bundled kuaixun-service 存在/不存在
|
||
- site-packages 发现（python* 目录 → site-packages）
|
||
- 无 .venv/lib 时返回 None
|
||
- 无 python* 目录时返回 None
|
||
- 无 site-packages 子目录时返回 None
|
||
"""
|
||
|
||
def test_bundled_kxs_exists_with_venv(self, tmp_path, monkeypatch):
|
||
"""bundled kuaixun-service 存在且有完整 venv 结构."""
|
||
import common
|
||
|
||
# resources_dir = Path(BIN_DIR).parent, 所以 BIN_DIR = tmp_path/bin
|
||
bin_dir = tmp_path / "bin"
|
||
bin_dir.mkdir()
|
||
kxs = tmp_path / "kuaixun-service"
|
||
# 版本无关：用 common 中统一的目标版本常量，避免硬编码 Python 版本
|
||
import common
|
||
sp = kxs / ".venv" / "lib" / f"python{common.PYTHON_TARGET_VERSION}" / "site-packages"
|
||
sp.mkdir(parents=True)
|
||
|
||
monkeypatch.setattr("common.BIN_DIR", str(bin_dir))
|
||
|
||
bundled_kxs, bundled_sp = common._resolve_bundled_paths()
|
||
assert bundled_kxs == str(kxs)
|
||
assert bundled_sp == str(sp)
|
||
|
||
def test_bundled_kxs_not_exists(self, tmp_path, monkeypatch):
|
||
"""bundled kuaixun-service 不存在时返回 None."""
|
||
import common
|
||
|
||
bin_dir = tmp_path / "bin"
|
||
bin_dir.mkdir()
|
||
monkeypatch.setattr("common.BIN_DIR", str(bin_dir))
|
||
|
||
bundled_kxs, bundled_sp = common._resolve_bundled_paths()
|
||
assert bundled_kxs is None
|
||
assert bundled_sp is None
|
||
|
||
def test_bundled_kxs_exists_no_venv(self, tmp_path, monkeypatch):
|
||
"""bundled kuaixun-service 存在但无 .venv/lib 目录."""
|
||
import common
|
||
|
||
bin_dir = tmp_path / "bin"
|
||
bin_dir.mkdir()
|
||
kxs = tmp_path / "kuaixun-service"
|
||
kxs.mkdir(parents=True)
|
||
|
||
monkeypatch.setattr("common.BIN_DIR", str(bin_dir))
|
||
|
||
bundled_kxs, bundled_sp = common._resolve_bundled_paths()
|
||
assert bundled_kxs == str(kxs)
|
||
assert bundled_sp is None
|
||
|
||
def test_bundled_kxs_has_venv_no_python_dir(self, tmp_path, monkeypatch):
|
||
"""bundled kxs 有 .venv/lib 但无 python* 子目录."""
|
||
import common
|
||
|
||
bin_dir = tmp_path / "bin"
|
||
bin_dir.mkdir()
|
||
kxs = tmp_path / "kuaixun-service"
|
||
lib_dir = kxs / ".venv" / "lib"
|
||
lib_dir.mkdir(parents=True)
|
||
|
||
monkeypatch.setattr("common.BIN_DIR", str(bin_dir))
|
||
|
||
bundled_kxs, bundled_sp = common._resolve_bundled_paths()
|
||
assert bundled_kxs == str(kxs)
|
||
assert bundled_sp is None
|
||
|
||
def test_bundled_kxs_has_python_dir_no_site_packages(self, tmp_path, monkeypatch):
|
||
"""bundled kxs 有 python* 目录但无 site-packages 子目录."""
|
||
import common
|
||
|
||
bin_dir = tmp_path / "bin"
|
||
bin_dir.mkdir()
|
||
kxs = tmp_path / "kuaixun-service"
|
||
py_dir = kxs / ".venv" / "lib" / "python3.10"
|
||
py_dir.mkdir(parents=True)
|
||
|
||
monkeypatch.setattr("common.BIN_DIR", str(bin_dir))
|
||
|
||
bundled_kxs, bundled_sp = common._resolve_bundled_paths()
|
||
assert bundled_kxs == str(kxs)
|
||
assert bundled_sp is None
|
||
|
||
def test_multiple_python_dirs_picks_first(self, tmp_path, monkeypatch):
|
||
"""多个 python* 目录时只取第一个找到的 site-packages."""
|
||
import common
|
||
|
||
bin_dir = tmp_path / "bin"
|
||
bin_dir.mkdir()
|
||
kxs = tmp_path / "kuaixun-service"
|
||
# 模拟版本迁移残留：旧版 3.10 目录仍在，当前目标是 3.12
|
||
sp1 = kxs / ".venv" / "lib" / "python3.10" / "site-packages"
|
||
sp1.mkdir(parents=True)
|
||
sp2 = kxs / ".venv" / "lib" / f"python3.12" / "site-packages"
|
||
sp2.mkdir(parents=True)
|
||
|
||
monkeypatch.setattr("common.BIN_DIR", str(bin_dir))
|
||
|
||
bundled_kxs, bundled_sp = common._resolve_bundled_paths()
|
||
assert bundled_kxs == str(kxs)
|
||
assert bundled_sp is not None
|
||
assert "python3" in bundled_sp
|
||
|
||
|
||
# ── setup_kuaixun_paths ─────────────────────────────────────
|
||
|
||
|
||
class TestSetupKuaixunPaths:
|
||
"""setup_kuaixun_paths 路径注入测试.
|
||
|
||
覆盖：
|
||
- bundled 路径存在时优先使用 bundled 路径
|
||
- bundled 路径不存在时回退到外部 KX_HOME/KXS_HOME
|
||
- BIN_DIR 始终被加入 sys.path
|
||
"""
|
||
|
||
def test_adds_bin_dir_to_sys_path(self, monkeypatch):
|
||
"""BIN_DIR 始终被加入 sys.path."""
|
||
import common
|
||
|
||
monkeypatch.setattr("common.BIN_DIR", "/fake/bin")
|
||
monkeypatch.setattr("common.BUNDLED_KXS_HOME", None)
|
||
monkeypatch.setattr("common.BUNDLED_VENV_SITE", None)
|
||
|
||
common.setup_kuaixun_paths()
|
||
assert "/fake/bin" in sys.path
|
||
|
||
def test_bundled_paths_used_when_available(self, monkeypatch, tmp_path):
|
||
"""bundled 路径存在时优先使用."""
|
||
import common
|
||
|
||
bundled_kxs = str(tmp_path / "bundled-kxs")
|
||
bundled_site = str(tmp_path / "bundled-site")
|
||
monkeypatch.setattr("common.BIN_DIR", "/fake/bin")
|
||
monkeypatch.setattr("common.BUNDLED_KXS_HOME", bundled_kxs)
|
||
monkeypatch.setattr("common.BUNDLED_VENV_SITE", bundled_site)
|
||
|
||
common.setup_kuaixun_paths()
|
||
assert bundled_kxs in sys.path
|
||
assert bundled_site in sys.path
|
||
|
||
def test_fallback_paths_when_no_bundled(self, monkeypatch):
|
||
"""bundled 不存在时回退到外部 KX_HOME 和 KXS_HOME."""
|
||
import common
|
||
|
||
monkeypatch.setattr("common.BIN_DIR", "/fake/bin")
|
||
monkeypatch.setattr("common.BUNDLED_KXS_HOME", None)
|
||
monkeypatch.setattr("common.BUNDLED_VENV_SITE", None)
|
||
monkeypatch.setattr("common.KX_HOME", "/fake/kx-home")
|
||
monkeypatch.setattr("common.KXS_HOME", "/fake/kxs-home")
|
||
monkeypatch.setattr("common.KX_VENV", "/fake/kx-venv")
|
||
|
||
common.setup_kuaixun_paths()
|
||
assert "/fake/kx-home" in sys.path
|
||
assert "/fake/kxs-home" in sys.path
|
||
assert "/fake/kx-venv" in sys.path
|