FinDataHub
f112e9d277
## 问题
backfill_stock_daily.py 把部署路径写死在 sys.path.insert 里：
sys.path.insert(0, "/Applications/FinDataHub.app/Contents/Resources/akshare_fetcher/src")
macOS 现网能跑，换平台或换安装位置即 ImportError。这是待办清单里长期挂着的 L2。
## 为什么不在脚本里写 try/except 双路径
那样只解决 bundle 形态，源码树运行时仍走 hardcode。正确做法是让「资源在哪」
由 paths.py 统一回答——它已经是路径的唯一事实来源。
## 新增三个函数
- resources_root()：bundle → <App>/Contents/Resources；源码树 → 仓库根
- service_dir(name)：<root>/<name> 或 <root>/Services/<name>
- service_src_dir(name)：service_dir(name)/"src"，可直接塞进 sys.path
**判据是内容而非层级**：仓库根下有 Services/，bundle 的资源目录下没有
（build-app.sh 把各服务平铺拷进 Resources/）。首版按「两级往上」硬算，
源码树里解析成 <repo>/Scripts——被 test_real_repo_layout_resolves 当场抓出。
## 验证（两形态实跑）
源码树：<repo>/Services/akshare_fetcher/src 存在 ✓
bundle：/Applications/…/Resources/akshare_fetcher/src 存在 ✓
import akshare_fetcher 成功 ✓
bundle 侧解析结果与旧硬编码**逐字一致**，故行为零漂移。
## 回归
TestServiceLayout 6 条：bundle / 源码树 / 两者共存时优先 bundle /
都不存在时返回 bundle 形态（错误信息才能指向期望位置）/ src 后缀 / 真实仓库解析。
scripts 257 → 263。
272 lines
11 KiB
Python
272 lines
11 KiB
Python
# coding=utf-8
|
||
"""paths.py / process_utils.py 跨平台抽象层测试.
|
||
|
||
本套件的第一职责是**防止迁移期行为漂移**：macOS 上新抽象必须解析出
|
||
与旧硬编码**逐字相同**的路径。历史上采集器 PID 文件、心跳文件的位置
|
||
被 launchd plist、Swift 端探测逻辑、运行中进程三处同时引用，
|
||
任何一处变化都会表现为"采集器在跑、App 却报未运行"。
|
||
|
||
第二职责是覆盖 Windows 分支的路径语义（不依赖真的跑在 Windows 上，
|
||
用 monkeypatch 切 ``IS_WINDOWS`` / ``sys.platform``）。
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
_bin_dir = os.path.join(os.path.dirname(__file__), "..")
|
||
if _bin_dir not in sys.path:
|
||
sys.path.insert(0, _bin_dir)
|
||
|
||
import paths # noqa: E402
|
||
import process_utils # noqa: E402
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════
|
||
# paths —— macOS 等价性（回归保护，最关键）
|
||
# ══════════════════════════════════════════════════════════
|
||
|
||
@pytest.mark.skipif(sys.platform != "darwin", reason="仅在 macOS 上校验等价性")
|
||
class TestMacOSEquivalence:
|
||
"""断言与迁移前硬编码完全一致。改这些断言前先想清楚兼容性。"""
|
||
|
||
def test_pid_file_matches_legacy_literal(self):
|
||
# 旧写法：PID_FILE = "/tmp/findatahub-newsflash.pid"
|
||
assert paths.pid_file("newsflash") == "/tmp/findatahub-newsflash.pid"
|
||
|
||
def test_backend_pid_file(self):
|
||
# 旧写法："/tmp/findatahub-newsflash-backend.pid"
|
||
assert paths.pid_file("newsflash-backend") == "/tmp/findatahub-newsflash-backend.pid"
|
||
|
||
def test_heartbeat_file(self):
|
||
# 旧写法："/tmp/findatahub-newsflash.heartbeat"
|
||
assert paths.heartbeat_file("newsflash") == "/tmp/findatahub-newsflash.heartbeat"
|
||
|
||
def test_temp_dir_is_tmp_not_gettempdir(self):
|
||
"""必须返回 /tmp，而不是 $TMPDIR。
|
||
|
||
tempfile.gettempdir() 在 macOS 返回 /var/folders/xx/.../T，
|
||
与现网 PID 文件所在 /tmp 不是同一个目录——混用即事故。
|
||
"""
|
||
assert paths.temp_dir() == Path("/tmp")
|
||
import tempfile
|
||
# 明确记录两者不同，避免后人"顺手"改回 gettempdir
|
||
assert Path(tempfile.gettempdir()) != paths.temp_dir()
|
||
|
||
def test_log_root(self):
|
||
assert paths.log_root() == Path.home() / "Library/Logs/FinDataHub"
|
||
|
||
def test_log_dir_with_sub(self):
|
||
assert paths.log_dir("newsflash") == (
|
||
Path.home() / "Library/Logs/FinDataHub/newsflash"
|
||
)
|
||
|
||
def test_app_support_dir(self):
|
||
assert paths.app_support_dir() == (
|
||
Path.home() / "Library/Application Support/FinDataHub"
|
||
)
|
||
|
||
def test_backups_dir(self):
|
||
assert paths.backups_dir() == (
|
||
Path.home() / "Library/Application Support/FinDataHub/backups"
|
||
)
|
||
|
||
def test_user_defaults_plist(self):
|
||
assert paths.user_defaults_plist() == (
|
||
Path.home() / "Library/Preferences/com.findatahub.app.plist"
|
||
)
|
||
|
||
def test_config_dir(self):
|
||
assert paths.config_dir() == Path.home() / ".config/findatahub"
|
||
|
||
def test_env_file(self):
|
||
assert paths.env_file() == Path.home() / ".config/findatahub/env"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════
|
||
# paths —— Windows 分支
|
||
# ══════════════════════════════════════════════════════════
|
||
|
||
class TestWindowsPaths:
|
||
@pytest.fixture
|
||
def win(self, monkeypatch):
|
||
monkeypatch.setattr(paths, "IS_WINDOWS", True)
|
||
monkeypatch.setattr(paths, "IS_MACOS", False)
|
||
monkeypatch.setenv("LOCALAPPDATA", r"C:\Users\tester\AppData\Local")
|
||
monkeypatch.setenv("APPDATA", r"C:\Users\tester\AppData\Roaming")
|
||
return paths
|
||
|
||
def test_log_root_under_localappdata(self, win):
|
||
assert win.log_root() == Path(r"C:\Users\tester\AppData\Local/FinDataHub/logs")
|
||
|
||
def test_app_support_under_localappdata(self, win):
|
||
assert win.app_support_dir() == Path(r"C:\Users\tester\AppData\Local/FinDataHub")
|
||
|
||
def test_config_under_appdata(self, win):
|
||
assert win.config_dir() == Path(r"C:\Users\tester\AppData\Roaming/FinDataHub")
|
||
|
||
def test_no_plist_on_windows(self, win):
|
||
"""Windows 无 plist——调用方必须处理 None，不能假设路径存在."""
|
||
assert win.user_defaults_plist() is None
|
||
|
||
def test_pid_file_uses_system_temp(self, win):
|
||
p = win.pid_file("newsflash")
|
||
assert p.endswith("findatahub-newsflash.pid")
|
||
assert "/tmp/" not in p
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════
|
||
# process_utils
|
||
# ══════════════════════════════════════════════════════════
|
||
|
||
class TestDetach:
|
||
def test_posix_uses_start_new_session(self):
|
||
if process_utils.IS_WINDOWS:
|
||
pytest.skip("仅 POSIX")
|
||
assert process_utils.detach_kwargs() == {"start_new_session": True}
|
||
|
||
def test_windows_uses_creationflags(self, monkeypatch):
|
||
monkeypatch.setattr(process_utils, "IS_WINDOWS", True)
|
||
kw = process_utils.detach_kwargs()
|
||
assert "creationflags" in kw
|
||
assert "start_new_session" not in kw
|
||
# CREATE_NEW_PROCESS_GROUP | DETACHED_PROCESS
|
||
assert kw["creationflags"] & 0x00000200
|
||
assert kw["creationflags"] & 0x00000008
|
||
|
||
def test_hidden_window_noop_on_posix(self):
|
||
if process_utils.IS_WINDOWS:
|
||
pytest.skip("仅 POSIX")
|
||
assert process_utils.hidden_window_kwargs() == {}
|
||
|
||
|
||
class TestIsRunning:
|
||
def test_current_process_alive(self):
|
||
assert process_utils.is_running(os.getpid()) is True
|
||
|
||
def test_nonexistent_pid(self):
|
||
# 取一个几乎不可能存在的 PID
|
||
assert process_utils.is_running(999_999) is False
|
||
|
||
@pytest.mark.parametrize("bad", [0, -1, -999])
|
||
def test_invalid_pid(self, bad):
|
||
"""非法 PID 直接判死，不得交给 os.kill 触发异常或误杀进程组."""
|
||
assert process_utils.is_running(bad) is False
|
||
|
||
|
||
class TestTerminate:
|
||
def test_invalid_pid_noop(self):
|
||
assert process_utils.terminate_pid(0) is False
|
||
assert process_utils.terminate_pid(-1) is False
|
||
|
||
def test_terminate_nonexistent_returns_false(self):
|
||
assert process_utils.terminate_pid(999_999) is False
|
||
|
||
|
||
class TestSignals:
|
||
def test_posix_has_core_signals(self):
|
||
if process_utils.IS_WINDOWS:
|
||
pytest.skip("仅 POSIX")
|
||
sigs = process_utils.available_signals()
|
||
assert "SIGTERM" in sigs
|
||
assert "SIGINT" in sigs
|
||
# SIGHUP 在 Windows 上不存在，POSIX 上应存在
|
||
assert "SIGHUP" in sigs
|
||
|
||
def test_register_skips_missing_signals(self):
|
||
"""平台不存在的信号必须跳过而不是 AttributeError 崩溃."""
|
||
called = []
|
||
ok = process_utils.register_signal_handler(
|
||
lambda s, f: called.append(s),
|
||
names=("SIGTERM", "SIG_THIS_DOES_NOT_EXIST"),
|
||
)
|
||
assert "SIGTERM" in ok
|
||
assert "SIG_THIS_DOES_NOT_EXIST" not in ok
|
||
|
||
def test_register_returns_registered_names(self):
|
||
original = process_utils.signal.getsignal(process_utils.signal.SIGINT)
|
||
try:
|
||
names = process_utils.register_signal_handler(
|
||
lambda s, f: None, names=("SIGINT",)
|
||
)
|
||
assert names == ["SIGINT"]
|
||
finally:
|
||
process_utils.signal.signal(process_utils.signal.SIGINT, original)
|
||
|
||
|
||
class TestUtf8Env:
|
||
def test_sets_utf8_vars(self):
|
||
env = process_utils.utf8_subprocess_env({"PATH": "/usr/bin"})
|
||
assert env["PYTHONUTF8"] == "1"
|
||
assert env["PYTHONIOENCODING"] == "utf-8"
|
||
assert env["PATH"] == "/usr/bin"
|
||
|
||
def test_does_not_mutate_input(self):
|
||
base = {"PATH": "/usr/bin"}
|
||
process_utils.utf8_subprocess_env(base)
|
||
assert "PYTHONUTF8" not in base, "必须返回副本，否则污染调用方环境"
|
||
|
||
def test_defaults_to_os_environ_copy(self):
|
||
env = process_utils.utf8_subprocess_env()
|
||
assert env is not os.environ
|
||
assert env["PYTHONUTF8"] == "1"
|
||
|
||
|
||
class TestServiceLayout:
|
||
"""bundle / 源码树两种布局的资源定位。
|
||
|
||
背景：``backfill_stock_daily.py`` 曾把
|
||
``/Applications/FinDataHub.app/Contents/Resources/akshare_fetcher/src``
|
||
写死在 ``sys.path.insert`` 里——macOS 现网能跑，但换平台或换安装位置即 ImportError。
|
||
``service_dir()`` 必须同时认两种布局。
|
||
"""
|
||
|
||
def _fake_layout(self, tmp_path, monkeypatch, *, bundle: bool, tree: bool):
|
||
"""构造一个假的 bin 目录，并把 paths.__file__ 指过去."""
|
||
root = tmp_path / ("Resources" if bundle else "repo")
|
||
bin_dir = root / "bin"
|
||
bin_dir.mkdir(parents=True)
|
||
|
||
if bundle:
|
||
(root / "akshare_fetcher" / "src").mkdir(parents=True)
|
||
if tree:
|
||
(root / "Services" / "akshare_fetcher" / "src").mkdir(parents=True)
|
||
|
||
monkeypatch.setattr(paths, "__file__", str(bin_dir / "paths.py"))
|
||
return root
|
||
|
||
def test_bundle_layout(self, tmp_path, monkeypatch):
|
||
root = self._fake_layout(tmp_path, monkeypatch, bundle=True, tree=False)
|
||
assert paths.service_dir("akshare_fetcher") == root / "akshare_fetcher"
|
||
|
||
def test_source_tree_layout(self, tmp_path, monkeypatch):
|
||
root = self._fake_layout(tmp_path, monkeypatch, bundle=False, tree=True)
|
||
assert paths.service_dir("akshare_fetcher") == root / "Services" / "akshare_fetcher"
|
||
|
||
def test_bundle_wins_when_both_exist(self, tmp_path, monkeypatch):
|
||
"""两种布局同时存在时优先 bundle——部署位才是生产真相."""
|
||
root = self._fake_layout(tmp_path, monkeypatch, bundle=True, tree=True)
|
||
assert paths.service_dir("akshare_fetcher") == root / "akshare_fetcher"
|
||
|
||
def test_missing_returns_bundle_shape(self, tmp_path, monkeypatch):
|
||
"""都不存在时返回 bundle 形态，错误信息才能指向期望位置."""
|
||
root = self._fake_layout(tmp_path, monkeypatch, bundle=False, tree=False)
|
||
assert paths.service_dir("no_such_svc") == root / "no_such_svc"
|
||
|
||
def test_src_dir_appends_src(self, tmp_path, monkeypatch):
|
||
root = self._fake_layout(tmp_path, monkeypatch, bundle=True, tree=False)
|
||
assert paths.service_src_dir("akshare_fetcher") == root / "akshare_fetcher" / "src"
|
||
assert paths.service_src_dir("akshare_fetcher").is_dir()
|
||
|
||
def test_real_repo_layout_resolves(self):
|
||
"""真实仓库下必须解析到 Services/akshare_fetcher 且 src 存在.
|
||
|
||
这条是「部署脚本漏拷服务目录」的第一道信号：源码树都解析不出来，
|
||
bundle 里更不会有。
|
||
"""
|
||
svc = paths.service_dir("akshare_fetcher")
|
||
assert (svc / "src").is_dir(), f"源码树未解析到 src 目录: {svc}"
|
||
assert svc.name == "akshare_fetcher"
|