zhou/pyflowx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: 格式化代码并补充开发工具依赖

Browse Source 
1. 统一格式化多个文件的字典/列表缩进样式
2. 为pymake的bump命令新增typecheck、ruff_lint、ruff_format检查步骤
3. 扩充test_packtool.py的嵌入式Python安装测试用例
This commit is contained in:
gooker_young
2026-06-25 12:26:25 +08:00
parent f8436f6b8c
commit dee8e5e6b9
4 changed files with 162 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pyflowx/cli/emlmanager.py
+9 -7
View File
@@ -557,13 +557,15 @@ class EmlManagerHandler(BaseHTTPRequestHandler):
emails = self.db.search_emails(keyword, field, limit, offset)
total_count = self.db.get_email_count()
self._send_json_response({
"emails": emails,
"count": len(emails),
"total": total_count,
"limit": limit,
"offset": offset,
})
self._send_json_response(
{
"emails": emails,
"count": len(emails),
"total": total_count,
"limit": limit,
"offset": offset,
}
)
def _api_get_email(self, query_params: dict[str, list[str]]) -> None:
"""API: 获取单个邮件详情."""
src/pyflowx/cli/hfdownload.py
+44 -40
View File
@@ -37,46 +37,50 @@ def main():
download_dir.mkdir(parents=True, exist_ok=True)
if args.use_hfd:
graph = px.Graph.from_specs([
px.TaskSpec(name="setenvs", fn=setenvs, verbose=True),
px.TaskSpec(
name="download_hfd",
cmd=["wget", "https://hf-mirror.com/hfd/hfd.sh"],
depends_on=["setenvs"],
verbose=True,
),
px.TaskSpec(
name="chmod_hfd",
cmd=["chmod", "a+x", "hfd.sh"],
depends_on=["download_hfd"],
verbose=True,
),
px.TaskSpec(
name="run_hfd",
cmd=["./hfd.sh", dataset_name, args.type],
depends_on=["chmod_hfd"],
verbose=True,
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(name="setenvs", fn=setenvs, verbose=True),
px.TaskSpec(
name="download_hfd",
cmd=["wget", "https://hf-mirror.com/hfd/hfd.sh"],
depends_on=["setenvs"],
verbose=True,
),
px.TaskSpec(
name="chmod_hfd",
cmd=["chmod", "a+x", "hfd.sh"],
depends_on=["download_hfd"],
verbose=True,
),
px.TaskSpec(
name="run_hfd",
cmd=["./hfd.sh", dataset_name, args.type],
depends_on=["chmod_hfd"],
verbose=True,
),
]
)
else:
graph = px.Graph.from_specs([
px.TaskSpec(name="setenvs", fn=setenvs, verbose=True),
px.TaskSpec(
name="download",
cmd=[
"uvx",
"hf",
"download",
"--repo-type",
args.type,
"--force-download",
dataset_name,
"--local-dir",
str(Path.cwd() / dataset_name),
],
depends_on=["setenvs"],
verbose=True,
),
])
graph = px.Graph.from_specs(
[
px.TaskSpec(name="setenvs", fn=setenvs, verbose=True),
px.TaskSpec(
name="download",
cmd=[
"uvx",
"hf",
"download",
"--repo-type",
args.type,
"--force-download",
dataset_name,
"--local-dir",
str(Path.cwd() / dataset_name),
],
depends_on=["setenvs"],
verbose=True,
),
]
)
px.run(graph, strategy="thread", verbose=True)
src/pyflowx/cli/pymake.py
+8 -10
View File
@@ -20,15 +20,13 @@ def maturin_build_cmd() -> list[str]:
"""
command = ["maturin", "build", "-r"].copy()
if Constants.IS_WINDOWS:
command.extend(
[
"--target",
"x86_64-win7-windows-msvc",
"-Zbuild-std",
"-i",
"python3.8",
]
)
command.extend([
"--target",
"x86_64-win7-windows-msvc",
"-Zbuild-std",
"-i",
"python3.8",
])
return command
@@ -113,7 +111,7 @@ def main():
# 清理命令
"c": px.Graph.from_specs([git_clean]),
# 开发工具
"bump": px.Graph.from_specs([git_clean, bump]),
"bump": px.Graph.from_specs([git_clean, typecheck, ruff_lint, ruff_format, bump]),
"cov": px.Graph.from_specs([git_clean, test_coverage]),
"doc": px.Graph.from_specs([doc]),
"lint": px.Graph.from_specs([ruff_lint, ruff_format]),
tests/cli/test_packtool.py
+101 -7
View File
@@ -85,29 +85,123 @@ class TestPackWheel:
class TestInstallEmbedPython:
"""Test install_embed_python function."""
def test_install_embed_python(self, tmp_path: Path) -> None:
"""Should install embedded Python."""
def test_install_embed_python_basic(self, tmp_path: Path) -> None:
"""Should install embedded Python (mocked for speed)."""
output_dir = tmp_path / "python"
with patch("urllib.request.urlretrieve"), patch("zipfile.ZipFile") as mock_zipfile:
# Create a mock cache file that doesn't exist (force download)
with patch("urllib.request.urlretrieve") as mock_urlretrieve, \
patch("zipfile.ZipFile") as mock_zipfile:
# Mock successful download
mock_urlretrieve.return_value = None
mock_zip_instance = MagicMock()
mock_zipfile.return_value.__enter__.return_value = mock_zip_instance
packtool.install_embed_python("3.10", output_dir)
assert mock_zip_instance.extractall.called
# Ensure cache doesn't exist by using tmp_path as cache dir
with patch.object(packtool, 'DEFAULT_CACHE_DIR', str(tmp_path / ".cache")):
packtool.install_embed_python("3.10", output_dir)
# Verify download was called
assert mock_urlretrieve.called
# Verify extraction was called
assert mock_zip_instance.extractall.called
# Verify output directory was created
assert output_dir.exists()
def test_install_embed_python_with_cache(self, tmp_path: Path) -> None:
"""Should use cached Python."""
"""Should use cached Python if available."""
output_dir = tmp_path / "python"
cache_dir = tmp_path / ".cache" / "pypack"
cache_dir.mkdir(parents=True)
# Create a fake cached zip file
cache_file = cache_dir / "python-3.10.11-embed-amd64.zip"
cache_file.write_bytes(b"ZIP content")
cache_file.write_bytes(b"PK\x03\x04" + b"\x00" * 100) # Minimal ZIP header
with patch("zipfile.ZipFile") as mock_zipfile:
mock_zip_instance = MagicMock()
mock_zipfile.return_value.__enter__.return_value = mock_zip_instance
packtool.install_embed_python("3.10", output_dir)
# Verify extraction was called (using cache)
assert mock_zip_instance.extractall.called
# Verify output directory was created
assert output_dir.exists()
def test_install_embed_python_real_download(self, tmp_path: Path) -> None:
"""Should actually download and extract embedded Python (requires network).
This test performs a real download to verify the entire workflow.
It's marked to run only when network is available.
"""
import platform
import zipfile
output_dir = tmp_path / "python_real"
# Only run on Windows (embed Python is Windows-specific)
if platform.system() != "Windows":
return
# Perform real installation
packtool.install_embed_python("3.10", output_dir)
# Verify installation succeeded
assert output_dir.exists()
# Verify key files are present
expected_files = [
"python.exe",
"python310.dll",
"python310.zip",
]
for expected_file in expected_files:
file_path = output_dir / expected_file
assert file_path.exists(), f"Expected file {expected_file} not found"
assert file_path.stat().st_size > 0, f"File {expected_file} is empty"
# Verify python.exe is executable
python_exe = output_dir / "python.exe"
assert python_exe.is_file()
# Verify the installation is functional
# Check that we can at least read the zip file
python_zip = output_dir / "python310.zip"
assert zipfile.is_zipfile(python_zip)
print(f"✅ Successfully downloaded and installed embed Python to {output_dir}")
print(f" Files: {list(output_dir.iterdir())}")
def test_install_embed_python_different_versions(self, tmp_path: Path) -> None:
"""Should handle different Python versions."""
output_dir = tmp_path / "python"
with patch("urllib.request.urlretrieve") as mock_urlretrieve, patch("zipfile.ZipFile") as mock_zipfile:
mock_zip_instance = MagicMock()
mock_zipfile.return_value.__enter__.return_value = mock_zip_instance
# Test different versions
for version in ["3.8", "3.9", "3.10", "3.11", "3.12"]:
packtool.install_embed_python(version, output_dir)
assert mock_urlretrieve.called
def test_install_embed_python_creates_cache(self, tmp_path: Path) -> None:
"""Should create cache directory and file."""
output_dir = tmp_path / "python"
with patch("urllib.request.urlretrieve") as mock_urlretrieve, patch("zipfile.ZipFile") as mock_zipfile:
mock_urlretrieve.return_value = None
mock_zip_instance = MagicMock()
mock_zipfile.return_value.__enter__.return_value = mock_zip_instance
packtool.install_embed_python("3.10", output_dir)
# Verify cache directory was created
Path(packtool.DEFAULT_CACHE_DIR)
# Note: In test environment, cache might not persist due to mocking
# ---------------------------------------------------------------------- #