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 条。
570 lines
20 KiB
Python
570 lines
20 KiB
Python
"""sync_bridge.py 测试 — _pump_stream 流转发 + 子进程管理 + 辅助函数."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
from unittest.mock import patch
|
||
|
||
import pytest
|
||
|
||
# 将 Scripts/bin 加入 sys.path 以导入 sync_bridge
|
||
_script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
if _script_dir not in sys.path:
|
||
sys.path.insert(0, _script_dir)
|
||
|
||
import sync_bridge
|
||
|
||
|
||
# ── _pump_stream ─────────────────────────────────────────────
|
||
|
||
class _FakeSink:
|
||
"""记录写入的 sink，模拟 sys.stdout.buffer / sys.stderr.buffer."""
|
||
|
||
def __init__(self):
|
||
self.written: list[bytes] = []
|
||
self.flush_calls: list[int] = []
|
||
|
||
def write(self, data: bytes):
|
||
self.written.append(data)
|
||
|
||
def flush(self):
|
||
self.flush_calls.append(len(self.written))
|
||
|
||
|
||
class TestPumpStream:
|
||
def test_forwards_all_chunks(self):
|
||
"""_pump_stream 将 fd 中所有数据块转发到 sink."""
|
||
data = [b"hello\n", b"world\n", b""]
|
||
fd = _FakeReadFD(data)
|
||
sink = _FakeSink()
|
||
sync_bridge._pump_stream(fd, sink)
|
||
assert sink.written == [b"hello\n", b"world\n"]
|
||
|
||
def test_empty_fd_no_write(self):
|
||
"""fd 立即返回空字节时不写入任何内容."""
|
||
sink = _FakeSink()
|
||
sync_bridge._pump_stream(_FakeReadFD([b""]), sink)
|
||
assert sink.written == []
|
||
|
||
def test_os_error_breaks_loop(self):
|
||
"""OSError 时优雅退出（不传播异常）."""
|
||
sink = _FakeSink()
|
||
# os.read 在无效 fd 上抛出 OSError
|
||
sync_bridge._pump_stream(-1, sink)
|
||
assert sink.written == []
|
||
|
||
def test_large_chunk_forwarded(self):
|
||
"""大块数据（>8192 字节）正确转发."""
|
||
data = b"x" * 10000
|
||
fd = _FakeReadFD([data, b""])
|
||
sink = _FakeSink()
|
||
sync_bridge._pump_stream(fd, sink)
|
||
assert len(sink.written) == 1
|
||
assert sink.written[0] == data
|
||
|
||
def test_flush_called_after_each_write(self):
|
||
"""每次 write 后调用 flush，确保实时转发不被缓冲."""
|
||
data = [b"chunk1\n", b"chunk2\n", b""]
|
||
fd = _FakeReadFD(data)
|
||
sink = _FakeSink()
|
||
sync_bridge._pump_stream(fd, sink)
|
||
# 每次 write 后应紧跟一次 flush：written[0] 后 flush 时 written 长度为 1
|
||
assert len(sink.flush_calls) == 2
|
||
assert sink.flush_calls == [1, 2]
|
||
|
||
def test_multiple_chunks_each_flushed(self):
|
||
"""多个数据块时每个块后都 flush."""
|
||
data = [b"a\n", b"b\n", b"c\n", b"d\n", b""]
|
||
fd = _FakeReadFD(data)
|
||
sink = _FakeSink()
|
||
sync_bridge._pump_stream(fd, sink)
|
||
assert len(sink.flush_calls) == 4
|
||
assert sink.flush_calls == [1, 2, 3, 4]
|
||
|
||
|
||
# ── _find_project ────────────────────────────────────────────
|
||
|
||
class TestFindProject:
|
||
def test_returns_first_candidate_with_pyproject(self, tmp_path, monkeypatch):
|
||
"""返回第一个存在 pyproject.toml 的候选路径."""
|
||
# 创建包含 pyproject.toml 的目录
|
||
proj_dir = tmp_path / "test_proj"
|
||
proj_dir.mkdir()
|
||
(proj_dir / "pyproject.toml").write_text("")
|
||
|
||
# mock HOME 和 os.environ
|
||
monkeypatch.setattr(sync_bridge, "HOME", tmp_path)
|
||
monkeypatch.delenv("APP_RESOURCES", raising=False)
|
||
|
||
result = sync_bridge._find_project("test_proj")
|
||
assert result == proj_dir
|
||
|
||
def test_falls_back_to_first_candidate(self, tmp_path, monkeypatch):
|
||
"""所有候选都不存在时返回第一个候选（不崩溃）."""
|
||
monkeypatch.setattr(sync_bridge, "HOME", tmp_path)
|
||
monkeypatch.delenv("APP_RESOURCES", raising=False)
|
||
|
||
result = sync_bridge._find_project("nonexistent_proj")
|
||
# 返回第一个候选路径
|
||
assert result is not None
|
||
|
||
|
||
# ── load_token ───────────────────────────────────────────────
|
||
|
||
class TestLoadToken:
|
||
def test_reads_from_env_file(self, tmp_path, monkeypatch):
|
||
"""从 .env 文件读取 TUSHARE_TOKEN."""
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text("TUSHARE_TOKEN=test-token-123\nOTHER=value\n")
|
||
|
||
result = sync_bridge.load_token(tmp_path)
|
||
assert result == "test-token-123"
|
||
|
||
def test_quoted_token_stripped(self, tmp_path):
|
||
"""带引号的 token 值正确去除引号."""
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text('TUSHARE_TOKEN="quoted-token"\n')
|
||
|
||
result = sync_bridge.load_token(tmp_path)
|
||
assert result == "quoted-token"
|
||
|
||
def test_no_env_file_returns_env_var(self, tmp_path, monkeypatch):
|
||
"""无 .env 文件时回退到环境变量."""
|
||
monkeypatch.setenv("TUSHARE_TOKEN", "env-token")
|
||
result = sync_bridge.load_token(tmp_path)
|
||
assert result == "env-token"
|
||
|
||
def test_no_env_file_no_env_var_returns_empty(self, tmp_path, monkeypatch):
|
||
"""无 .env 且无环境变量时返回空字符串."""
|
||
monkeypatch.delenv("TUSHARE_TOKEN", raising=False)
|
||
result = sync_bridge.load_token(tmp_path)
|
||
assert result == ""
|
||
|
||
|
||
# ── run_akshare ───────────────────────────────────────────────
|
||
|
||
class TestRunAkshare:
|
||
def test_returns_exit_code(self, tmp_path, monkeypatch):
|
||
"""子进程正常退出时返回退出码."""
|
||
# 使用 echo 命令模拟子进程
|
||
monkeypatch.setattr(subprocess, "Popen", _make_fake_popen(returncode=0))
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
rc = sync_bridge.run_akshare(tmp_path, "daily", 4, "/usr/bin/python3")
|
||
assert rc == 0
|
||
|
||
def test_returns_nonzero_exit_code(self, tmp_path, monkeypatch):
|
||
"""子进程异常退出时返回非零退出码."""
|
||
monkeypatch.setattr(subprocess, "Popen", _make_fake_popen(returncode=1))
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
rc = sync_bridge.run_akshare(tmp_path, "daily", 4, "/usr/bin/python3")
|
||
assert rc == 1
|
||
|
||
def test_stdout_stderr_threads_started(self, tmp_path, monkeypatch):
|
||
"""stdout/stderr 转发线程被启动."""
|
||
threads_started = []
|
||
|
||
class _FakeThread:
|
||
def __init__(self, target, args, daemon=False):
|
||
self.target = target
|
||
self.args = args
|
||
self.daemon = daemon
|
||
threads_started.append(self)
|
||
|
||
def start(self):
|
||
pass
|
||
|
||
def join(self):
|
||
pass
|
||
|
||
monkeypatch.setattr(threading, "Thread", _FakeThread)
|
||
monkeypatch.setattr(subprocess, "Popen", _make_fake_popen(returncode=0))
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
sync_bridge.run_akshare(tmp_path, "daily", 4, "/usr/bin/python3")
|
||
assert len(threads_started) == 2
|
||
assert threads_started[0].daemon is True
|
||
assert threads_started[1].daemon is True
|
||
|
||
def test_popen_cmd_constructed_correctly(self, tmp_path, monkeypatch):
|
||
"""Popen cmd 参数正确构造：python -m akshare_fetcher.cli sync --type X --workers N."""
|
||
popen_calls = []
|
||
|
||
class _CapturePopen:
|
||
def __init__(self, cmd, **kwargs):
|
||
popen_calls.append((cmd, kwargs))
|
||
self._stdout = _FakeReadFD([b""])
|
||
self._stderr = _FakeReadFD([b""])
|
||
|
||
@property
|
||
def stdout(self):
|
||
return self._stdout
|
||
|
||
@property
|
||
def stderr(self):
|
||
return self._stderr
|
||
|
||
def wait(self):
|
||
return 0
|
||
|
||
monkeypatch.setattr(subprocess, "Popen", _CapturePopen)
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
sync_bridge.run_akshare(tmp_path, "weekly", 8, "/usr/bin/python3")
|
||
assert len(popen_calls) == 1
|
||
cmd = popen_calls[0][0]
|
||
assert cmd[0] == "/usr/bin/python3"
|
||
assert cmd[1] == "-m"
|
||
assert cmd[2] == "akshare_fetcher.cli"
|
||
assert cmd[3] == "sync"
|
||
assert "--type" in cmd
|
||
assert "weekly" in cmd
|
||
assert "--workers" in cmd
|
||
assert "8" in cmd
|
||
|
||
def test_popen_env_contains_required_vars(self, tmp_path, monkeypatch):
|
||
"""Popen env 包含 PYTHONPATH、AKSHARE_HOME、PYTHONUNBUFFERED."""
|
||
popen_calls = []
|
||
|
||
class _CapturePopen:
|
||
def __init__(self, cmd, **kwargs):
|
||
popen_calls.append((cmd, kwargs))
|
||
self._stdout = _FakeReadFD([b""])
|
||
self._stderr = _FakeReadFD([b""])
|
||
|
||
@property
|
||
def stdout(self):
|
||
return self._stdout
|
||
|
||
@property
|
||
def stderr(self):
|
||
return self._stderr
|
||
|
||
def wait(self):
|
||
return 0
|
||
|
||
monkeypatch.setattr(subprocess, "Popen", _CapturePopen)
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
sync_bridge.run_akshare(tmp_path, "daily", 4, "/usr/bin/python3")
|
||
env = popen_calls[0][1].get("env", {})
|
||
assert env["PYTHONPATH"] == f"{tmp_path}/src"
|
||
assert env["AKSHARE_HOME"] == str(tmp_path)
|
||
assert env["PYTHONUNBUFFERED"] == "1"
|
||
|
||
def test_popen_uses_pipes_for_streams(self, tmp_path, monkeypatch):
|
||
"""Popen 使用 PIPE 捕获 stdout 和 stderr."""
|
||
popen_calls = []
|
||
|
||
class _CapturePopen:
|
||
def __init__(self, cmd, **kwargs):
|
||
popen_calls.append((cmd, kwargs))
|
||
self._stdout = _FakeReadFD([b""])
|
||
self._stderr = _FakeReadFD([b""])
|
||
|
||
@property
|
||
def stdout(self):
|
||
return self._stdout
|
||
|
||
@property
|
||
def stderr(self):
|
||
return self._stderr
|
||
|
||
def wait(self):
|
||
return 0
|
||
|
||
monkeypatch.setattr(subprocess, "Popen", _CapturePopen)
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
sync_bridge.run_akshare(tmp_path, "monthly", 2, "/usr/bin/python3")
|
||
kwargs = popen_calls[0][1]
|
||
assert kwargs["stdout"] == subprocess.PIPE
|
||
assert kwargs["stderr"] == subprocess.PIPE
|
||
assert kwargs["cwd"] == str(tmp_path)
|
||
|
||
|
||
# ── run_tushare ──────────────────────────────────────────────
|
||
|
||
class TestRunTushare:
|
||
def test_no_token_returns_1(self, tmp_path, monkeypatch):
|
||
"""无 token 时立即返回 1."""
|
||
monkeypatch.setattr(sync_bridge, "load_token", lambda p: "")
|
||
rc = sync_bridge.run_tushare(tmp_path, "daily", 4, "/usr/bin/python3")
|
||
assert rc == 1
|
||
|
||
def test_batch_success_returns_0(self, tmp_path, monkeypatch):
|
||
"""所有 fetch type 成功时返回 0."""
|
||
monkeypatch.setattr(sync_bridge, "load_token", lambda p: "token")
|
||
monkeypatch.setattr(subprocess, "Popen", _make_fake_popen(returncode=0))
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
# daily batch 包含 ["daily", "market_sentiment"]
|
||
rc = sync_bridge.run_tushare(tmp_path, "daily", 4, "/usr/bin/python3")
|
||
assert rc == 0
|
||
|
||
def test_batch_partial_failure_returns_nonzero(self, tmp_path, monkeypatch):
|
||
"""任一 fetch type 失败时返回非零."""
|
||
monkeypatch.setattr(sync_bridge, "load_token", lambda p: "token")
|
||
|
||
class _FailThenOkPopen:
|
||
_call_count = 0
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
_FailThenOkPopen._call_count += 1
|
||
|
||
def wait(self):
|
||
if _FailThenOkPopen._call_count == 1:
|
||
return 1 # 第一个 fetch type 失败
|
||
return 0
|
||
|
||
@property
|
||
def stdout(self):
|
||
return _FakeReadFD([])
|
||
|
||
@property
|
||
def stderr(self):
|
||
return _FakeReadFD([])
|
||
|
||
monkeypatch.setattr(subprocess, "Popen", _FailThenOkPopen)
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
# 重置计数器
|
||
_FailThenOkPopen._call_count = 0
|
||
rc = sync_bridge.run_tushare(tmp_path, "daily", 4, "/usr/bin/python3")
|
||
assert rc == 1
|
||
|
||
def test_popen_env_contains_token(self, tmp_path, monkeypatch):
|
||
"""tushare Popen env 包含 TUSHARE_TOKEN."""
|
||
popen_calls = []
|
||
|
||
class _CapturePopen:
|
||
def __init__(self, cmd, **kwargs):
|
||
popen_calls.append((cmd, kwargs))
|
||
self._stdout = _FakeReadFD([b""])
|
||
self._stderr = _FakeReadFD([b""])
|
||
|
||
@property
|
||
def stdout(self):
|
||
return self._stdout
|
||
|
||
@property
|
||
def stderr(self):
|
||
return self._stderr
|
||
|
||
def wait(self):
|
||
return 0
|
||
|
||
monkeypatch.setattr(subprocess, "Popen", _CapturePopen)
|
||
monkeypatch.setattr(sync_bridge, "load_token", lambda p: "test-token-abc")
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
sync_bridge.run_tushare(tmp_path, "daily", 4, "/usr/bin/python3")
|
||
assert len(popen_calls) >= 1
|
||
env = popen_calls[0][1].get("env", {})
|
||
assert env["TUSHARE_TOKEN"] == "test-token-abc"
|
||
assert env["TUSHARE_HOME"] == str(tmp_path)
|
||
assert env["PYTHONUNBUFFERED"] == "1"
|
||
|
||
def test_weekly_batch_uses_correct_fetch_types(self, tmp_path, monkeypatch):
|
||
"""weekly batch 使用 [fundamental, static, event]."""
|
||
popen_calls = []
|
||
|
||
class _CapturePopen:
|
||
def __init__(self, cmd, **kwargs):
|
||
popen_calls.append((cmd, kwargs))
|
||
self._stdout = _FakeReadFD([b""])
|
||
self._stderr = _FakeReadFD([b""])
|
||
|
||
@property
|
||
def stdout(self):
|
||
return self._stdout
|
||
|
||
@property
|
||
def stderr(self):
|
||
return self._stderr
|
||
|
||
def wait(self):
|
||
return 0
|
||
|
||
monkeypatch.setattr(subprocess, "Popen", _CapturePopen)
|
||
monkeypatch.setattr(sync_bridge, "load_token", lambda p: "token")
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
sync_bridge.run_tushare(tmp_path, "weekly", 4, "/usr/bin/python3")
|
||
# weekly batch: fundamental, static, event = 3 Popen 调用
|
||
assert len(popen_calls) == 3
|
||
|
||
def test_unknown_sync_type_falls_back_to_daily(self, tmp_path, monkeypatch):
|
||
"""未知 sync_type 回退到 ["daily"]."""
|
||
popen_calls = []
|
||
|
||
class _CapturePopen:
|
||
def __init__(self, cmd, **kwargs):
|
||
popen_calls.append((cmd, kwargs))
|
||
self._stdout = _FakeReadFD([b""])
|
||
self._stderr = _FakeReadFD([b""])
|
||
|
||
@property
|
||
def stdout(self):
|
||
return self._stdout
|
||
|
||
@property
|
||
def stderr(self):
|
||
return self._stderr
|
||
|
||
def wait(self):
|
||
return 0
|
||
|
||
monkeypatch.setattr(subprocess, "Popen", _CapturePopen)
|
||
monkeypatch.setattr(sync_bridge, "load_token", lambda p: "token")
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
sync_bridge.run_tushare(tmp_path, "unknown_type", 4, "/usr/bin/python3")
|
||
assert len(popen_calls) == 1 # 仅 daily
|
||
|
||
def test_threads_started_per_fetch_type(self, tmp_path, monkeypatch):
|
||
"""每个 fetch type 启动 2 个 daemon 线程（stdout/stderr 转发）.
|
||
|
||
安全审计修复（4b62693）：run_tushare 从 capture_output=True（可能
|
||
缓冲满后死锁）改为 Popen + _pump_stream 线程转发。此测试验证
|
||
线程确实被创建且为 daemon，确保子进程输出不会被丢弃。
|
||
"""
|
||
threads_started = []
|
||
|
||
class _FakeThread:
|
||
def __init__(self, target, args, daemon=False):
|
||
self.target = target
|
||
self.args = args
|
||
self.daemon = daemon
|
||
threads_started.append(self)
|
||
|
||
def start(self):
|
||
pass
|
||
|
||
def join(self):
|
||
pass
|
||
|
||
monkeypatch.setattr(threading, "Thread", _FakeThread)
|
||
monkeypatch.setattr(sync_bridge, "load_token", lambda p: "token")
|
||
monkeypatch.setattr(subprocess, "Popen", _make_fake_popen(returncode=0))
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
# daily batch = ["daily", "market_sentiment"] → 2 fetch types × 2 threads = 4
|
||
sync_bridge.run_tushare(tmp_path, "daily", 4, "/usr/bin/python3")
|
||
assert len(threads_started) == 4
|
||
assert all(t.daemon is True for t in threads_started)
|
||
|
||
def test_popen_uses_pipes_not_capture_output(self, tmp_path, monkeypatch):
|
||
"""Popen 使用 stdout=PIPE/stderr=PIPE，不使用 capture_output.
|
||
|
||
capture_output=True 会缓冲子进程 stderr（tqdm 进度），
|
||
缓冲满后阻塞子进程写入导致死锁。必须使用 PIPE + 线程转发。
|
||
"""
|
||
popen_calls = []
|
||
|
||
class _CapturePopen:
|
||
def __init__(self, cmd, **kwargs):
|
||
popen_calls.append((cmd, kwargs))
|
||
self._stdout = _FakeReadFD([b""])
|
||
self._stderr = _FakeReadFD([b""])
|
||
|
||
@property
|
||
def stdout(self):
|
||
return self._stdout
|
||
|
||
@property
|
||
def stderr(self):
|
||
return self._stderr
|
||
|
||
def wait(self):
|
||
return 0
|
||
|
||
monkeypatch.setattr(subprocess, "Popen", _CapturePopen)
|
||
monkeypatch.setattr(sync_bridge, "load_token", lambda p: "token")
|
||
monkeypatch.setattr(sync_bridge, "find_python", lambda p: "/usr/bin/python3")
|
||
|
||
sync_bridge.run_tushare(tmp_path, "daily", 4, "/usr/bin/python3")
|
||
assert len(popen_calls) >= 1
|
||
for _, kwargs in popen_calls:
|
||
assert kwargs.get("stdout") == subprocess.PIPE
|
||
assert kwargs.get("stderr") == subprocess.PIPE
|
||
assert "capture_output" not in kwargs or not kwargs.get("capture_output")
|
||
|
||
|
||
# ── helpers ──────────────────────────────────────────────────
|
||
|
||
class _FakeReadFD:
|
||
"""模拟 os.read 可读取的文件描述符."""
|
||
|
||
def __init__(self, chunks: list[bytes]):
|
||
self._chunks = list(chunks)
|
||
self._idx = 0
|
||
|
||
def fileno(self):
|
||
return 999 # 虚拟 fd 号
|
||
|
||
|
||
def _mock_os_read_factory(chunks: list[bytes]):
|
||
"""创建 os.read mock，按顺序返回 chunks."""
|
||
|
||
def fake_read(fd, n):
|
||
# 仅在 fd 是我们创建的 FakeReadFD 时才拦截
|
||
if isinstance(fd, _FakeReadFD):
|
||
if fd._idx >= len(fd._chunks):
|
||
return b""
|
||
chunk = fd._chunks[fd._idx]
|
||
fd._idx += 1
|
||
return chunk
|
||
return os.read(fd, n)
|
||
|
||
return fake_read
|
||
|
||
|
||
def _make_fake_popen(returncode=0):
|
||
"""创建 fake subprocess.Popen 类."""
|
||
|
||
class FakePopen:
|
||
def __init__(self, cmd, **kwargs):
|
||
self.cmd = cmd
|
||
self.kwargs = kwargs
|
||
self._stdout = _FakeReadFD([b"ok\n", b""])
|
||
self._stderr = _FakeReadFD([b""])
|
||
|
||
@property
|
||
def stdout(self):
|
||
return self._stdout
|
||
|
||
@property
|
||
def stderr(self):
|
||
return self._stderr
|
||
|
||
def wait(self):
|
||
return returncode
|
||
|
||
return FakePopen
|
||
|
||
|
||
# 全局 fixture：用 mock os.read 替换 _pump_stream 中的 os.read 调用
|
||
@pytest.fixture(autouse=True)
|
||
def _mock_os_read(monkeypatch):
|
||
"""注入 fake os.read，使 _pump_stream 使用 FakeReadFD."""
|
||
orig_read = os.read
|
||
|
||
def fake_read(fd, n):
|
||
if isinstance(fd, _FakeReadFD):
|
||
if fd._idx >= len(fd._chunks):
|
||
return b""
|
||
chunk = fd._chunks[fd._idx]
|
||
fd._idx += 1
|
||
return chunk
|
||
try:
|
||
return orig_read(fd, n)
|
||
except OSError:
|
||
raise
|
||
|
||
monkeypatch.setattr(os, "read", fake_read) |