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

~bumpversion

Browse Source
This commit is contained in:
gooker_young
2026-06-25 23:21:42 +08:00
parent 0d6a78f320
commit f10f8d09a6
4 changed files with 351 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pyflowx/__init__.py
+1 -1
View File
@@ -84,7 +84,7 @@ from .runner import CliExitCode, CliRunner
from .storage import JSONBackend, MemoryBackend, StateBackend
from .task import TaskCmd, TaskEvent, TaskResult, TaskSpec, TaskStatus
__version__ = "0.1.13"
__version__ = "0.1.14"
__all__ = [
"IS_LINUX",
src/pyflowx/cli/bumpversion.py
+128 -72
View File
@@ -5,97 +5,153 @@
from __future__ import annotations
import subprocess
import argparse
import re
from pathlib import Path
from typing import Literal, get_args
import pyflowx as px
# ============================================================================
# 辅助函数
# ============================================================================
BumpVersionType = Literal["patch", "minor", "major"]
def bump_version(part: str = "patch", tag: bool = False, commit: bool = False) -> None:
"""递增版本号.
_VERSION_PATTERN = re.compile(
r"(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*)"
r"(?:-(?P<prerelease>(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?"
r"(?:\+(?P<buildmetadata>[0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?",
)
def bump_file_version(file_path: Path, part: BumpVersionType = "patch") -> str | None:
"""更新文件中的版本号.
Parameters
----------
part : str
file_path : Path
要更新的文件路径
part : BumpVersionType
版本部分: patch, minor, major
tag : bool
是否创建 Git 标签
commit : bool
是否提交更改
Returns
-------
str | None
更新后的新版本号，如果文件中未找到版本号则返回 None
"""
try:
subprocess.run(["bumpversion", part], check=True)
if commit:
subprocess.run(["git", "add", "."], check=True)
subprocess.run(["git", "commit", "-m", f"bump version {part}"], check=True)
if tag:
# 获取当前版本号
result = subprocess.run(
["git", "describe", "--tags", "--abbrev=0"],
check=True,
capture_output=True,
text=True,
)
version = result.stdout.strip() if result.returncode == 0 else f"v{part}"
subprocess.run(
["git", "tag", "-a", version, "-m", f"version {part}"],
check=True,
)
except FileNotFoundError:
print("未找到 bumpversion 工具，请先安装: pip install bumpversion")
content = file_path.read_text(encoding="utf-8")
except Exception as e:
print(f"读取文件 {file_path} 时出错: {e}")
raise
match = _VERSION_PATTERN.search(content)
if not match:
print(f"文件 {file_path} 中未找到版本号模式")
return None
major = int(match.group("major"))
minor = int(match.group("minor"))
patch = int(match.group("patch"))
# 计算新版本号
if part == "major":
new_major = major + 1
new_version_str = f"{new_major}.0.0"
elif part == "minor":
new_minor = minor + 1
new_version_str = f"{major}.{new_minor}.0"
else: # patch
new_patch = patch + 1
new_version_str = f"{major}.{minor}.{new_patch}"
content = content.replace(match.group(0), new_version_str)
def bump_version_alpha(part: str = "patch") -> None:
"""递增版本号并添加 alpha 预发布标识."""
try:
subprocess.run(["bumpversion", part, "--new-version", f"{part}-alpha"], check=True)
except FileNotFoundError:
print("未找到 bumpversion 工具，请先安装: pip install bumpversion")
file_path.write_text(content, encoding="utf-8")
except Exception as e:
print(f"更新文件 {file_path} 版本号时出错: {e}")
raise
# ============================================================================
# TaskSpec 定义
# ============================================================================
bump_patch: px.TaskSpec = px.TaskSpec("bump_patch", fn=lambda: bump_version("patch"))
bump_minor: px.TaskSpec = px.TaskSpec("bump_minor", fn=lambda: bump_version("minor"))
bump_major: px.TaskSpec = px.TaskSpec("bump_major", fn=lambda: bump_version("major"))
bump_patch_tag: px.TaskSpec = px.TaskSpec("bump_patch_tag", fn=lambda: bump_version("patch", tag=True))
bump_minor_tag: px.TaskSpec = px.TaskSpec("bump_minor_tag", fn=lambda: bump_version("minor", tag=True))
bump_major_tag: px.TaskSpec = px.TaskSpec("bump_major_tag", fn=lambda: bump_version("major", tag=True))
bump_patch_alpha: px.TaskSpec = px.TaskSpec("bump_patch_alpha", fn=lambda: bump_version_alpha("patch"))
# ============================================================================
# CLI Runner
# ============================================================================
return new_version_str
def main() -> None:
"""版本号管理工具主函数."""
runner = px.CliRunner(
strategy="thread",
description="BumpVersion - 版本号自动管理工具",
graphs={
# 递增补丁号 (1.0.0 -> 1.0.1)
"p": px.Graph.from_specs([bump_patch]),
# 递增次版本号 (1.0.0 -> 1.1.0)
"m": px.Graph.from_specs([bump_minor]),
# 递增主版本号 (1.0.0 -> 2.0.0)
"M": px.Graph.from_specs([bump_major]),
# 递增补丁号并创建标签
"pt": px.Graph.from_specs([bump_patch_tag]),
# 递增次版本号并创建标签
"mt": px.Graph.from_specs([bump_minor_tag]),
# 递增主版本号并创建标签
"Mt": px.Graph.from_specs([bump_major_tag]),
# 递增补丁号并添加 alpha 预发布标识
"pa": px.Graph.from_specs([bump_patch_alpha]),
},
parser = argparse.ArgumentParser(description="BumpVersion - 版本号自动管理工具")
parser.add_argument(
"part",
type=str,
nargs="?",
default="patch",
choices=get_args(BumpVersionType),
help=f"版本部分: {get_args(BumpVersionType)}",
)
runner.run_cli()
parser.add_argument(
"--no-tag",
action="store_true",
help="提交后不创建 git tag",
)
args = parser.parse_args()
part = args.part
# 搜索文件，排除常见的虚拟环境和缓存目录
ignore_dirs = {".venv", "venv", ".git", "__pycache__", ".tox", "node_modules", "build", "dist", ".eggs"}
all_files = set()
for pattern in ["__init__.py", "pyproject.toml"]:
for file in Path.cwd().rglob(pattern):
# 检查路径中是否包含需要忽略的目录
if not any(ignore_dir in file.parts for ignore_dir in ignore_dirs):
all_files.add(file)
if not all_files:
print("未找到包含版本号的文件")
return
print(f"找到 {len(all_files)} 个文件需要更新版本号")
for file in sorted(all_files):
print(f" - {file.relative_to(Path.cwd())}")
# 更新所有文件的版本号（使用顺序执行避免竞争条件）
# 使用相对于 cwd 的路径作为任务名，确保唯一性
graph = px.Graph.from_specs([
px.TaskSpec(
f"bump_{file.relative_to(Path.cwd())}".replace("\\", "_").replace("/", "_").replace(".", "_"),
fn=bump_file_version,
args=(file, part),
)
for file in all_files
])
report = px.run(graph, strategy="sequential")
# 收集新版本号（取第一个成功的结果）
new_version = None
for task_name in report:
result = report[task_name]
if result is not None:
new_version = result
break
if not new_version:
print("未能获取新版本号")
return
print(f"版本号已更新为: {new_version}")
# 提交修改
graph = px.Graph.from_specs([
px.TaskSpec("git_add", cmd=["git", "add", "."]),
px.TaskSpec(
"git_commit", cmd=["git", "commit", "-m", f"bump version to {new_version}"], depends_on=["git_add"]
),
])
px.run(graph, strategy="sequential")
# 创建 git tag
if not args.no_tag:
tag_name = f"v{new_version}"
graph = px.Graph.from_specs([
px.TaskSpec("git_tag", cmd=["git", "tag", "-a", tag_name, "-m", f"Release {tag_name}"]),
])
px.run(graph, strategy="sequential")
print(f"已创建标签: {tag_name}")
tests/cli/test_bumpversion.py
+221 -73
View File
@@ -2,105 +2,253 @@
from __future__ import annotations
from unittest.mock import MagicMock, patch
from pathlib import Path
import pytest
import pyflowx as px
from pyflowx.cli import bumpversion
# ---------------------------------------------------------------------- #
# bump_version
# bump_file_version
# ---------------------------------------------------------------------- #
class TestBumpVersion:
"""Test bump_version function."""
class TestBumpFileVersion:
"""Test bump_file_version function."""
def test_bump_version_patch(self) -> None:
"""Should bump patch version."""
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
bumpversion.bump_version("patch")
assert mock_run.called
def test_bump_patch_version(self, tmp_path: Path) -> None:
"""Should bump patch version correctly."""
test_file = tmp_path / "test.txt"
test_file.write_text("version = 1.2.3", encoding="utf-8")
def test_bump_version_minor(self) -> None:
"""Should bump minor version."""
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
bumpversion.bump_version("minor")
assert mock_run.called
result = bumpversion.bump_file_version(test_file, "patch")
def test_bump_version_major(self) -> None:
"""Should bump major version."""
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
bumpversion.bump_version("major")
assert mock_run.called
assert result == "1.2.4"
assert test_file.read_text(encoding="utf-8") == "version = 1.2.4"
def test_bump_version_with_tag(self) -> None:
"""Should bump version with tag."""
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0, stdout="v1.0.0")
bumpversion.bump_version("patch", tag=True)
assert mock_run.called
def test_bump_minor_version(self, tmp_path: Path) -> None:
"""Should bump minor version correctly."""
test_file = tmp_path / "test.txt"
test_file.write_text("version = 1.2.3", encoding="utf-8")
def test_bump_version_with_commit(self) -> None:
"""Should bump version with commit."""
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
bumpversion.bump_version("patch", commit=True)
assert mock_run.called
result = bumpversion.bump_file_version(test_file, "minor")
def test_bump_version_file_not_found(self) -> None:
"""Should handle FileNotFoundError."""
with patch("subprocess.run", side_effect=FileNotFoundError), pytest.raises(FileNotFoundError):
bumpversion.bump_version("patch")
assert result == "1.3.0"
assert test_file.read_text(encoding="utf-8") == "version = 1.3.0"
def test_bump_major_version(self, tmp_path: Path) -> None:
"""Should bump major version correctly."""
test_file = tmp_path / "test.txt"
test_file.write_text("version = 1.2.3", encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "major")
assert result == "2.0.0"
assert test_file.read_text(encoding="utf-8") == "version = 2.0.0"
def test_version_pattern_with_prerelease(self, tmp_path: Path) -> None:
"""Should handle version with prerelease suffix."""
test_file = tmp_path / "test.txt"
test_file.write_text("version = 1.2.3-alpha.1", encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "patch")
assert result == "1.2.4"
# 预发布版本应该被清除
content = test_file.read_text(encoding="utf-8")
assert "alpha" not in content
def test_version_pattern_with_build_metadata(self, tmp_path: Path) -> None:
"""Should handle version with build metadata."""
test_file = tmp_path / "test.txt"
test_file.write_text("version = 1.2.3+build.123", encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "patch")
assert result == "1.2.4"
# 构建元数据应该被清除
content = test_file.read_text(encoding="utf-8")
assert "build" not in content
def test_no_version_found(self, tmp_path: Path, capsys) -> None:
"""Should return None when no version pattern found."""
test_file = tmp_path / "test.txt"
test_file.write_text("no version here", encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "patch")
assert result is None
captured = capsys.readouterr()
assert "未找到版本号模式" in captured.out
def test_utf8_encoding(self, tmp_path: Path) -> None:
"""Should handle UTF-8 encoded files correctly."""
test_file = tmp_path / "test.txt"
test_file.write_text("版本 = 1.2.3", encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "patch")
assert result == "1.2.4"
assert test_file.read_text(encoding="utf-8") == "版本 = 1.2.4"
def test_pyproject_toml_format(self, tmp_path: Path) -> None:
"""Should handle pyproject.toml format correctly."""
test_file = tmp_path / "pyproject.toml"
content = """
[project]
name = "test"
version = "0.1.0"
description = "Test project"
"""
test_file.write_text(content, encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "minor")
assert result == "0.2.0"
updated = test_file.read_text(encoding="utf-8")
assert 'version = "0.2.0"' in updated
assert 'name = "test"' in updated
def test_init_py_format(self, tmp_path: Path) -> None:
"""Should handle __init__.py format correctly."""
test_file = tmp_path / "__init__.py"
content = '''"""Package info."""
__version__ = "1.0.0"
'''
test_file.write_text(content, encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "major")
assert result == "2.0.0"
updated = test_file.read_text(encoding="utf-8")
assert '__version__ = "2.0.0"' in updated
def test_multiple_versions_in_file(self, tmp_path: Path) -> None:
"""Should only bump the first version occurrence."""
test_file = tmp_path / "test.txt"
test_file.write_text("version1 = 1.0.0\nversion2 = 2.0.0", encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "patch")
assert result == "1.0.1"
content = test_file.read_text(encoding="utf-8")
assert "version1 = 1.0.1" in content
assert "version2 = 2.0.0" in content
def test_file_read_error(self, tmp_path: Path, capsys) -> None:
"""Should handle file read errors."""
# 创建一个目录而不是文件
test_file = tmp_path / "test_dir"
test_file.mkdir()
with pytest.raises(Exception): # noqa: B017
bumpversion.bump_file_version(test_file, "patch")
def test_file_write_error(self, tmp_path: Path, capsys) -> None:
"""Should handle file write errors."""
# 在只读目录中创建文件（这个测试在某些系统上可能不适用）
test_file = tmp_path / "readonly.txt"
test_file.write_text("version = 1.0.0", encoding="utf-8")
# 设置为只读
test_file.chmod(0o444)
try:
with pytest.raises(Exception):
bumpversion.bump_file_version(test_file, "patch")
finally:
# 恢复权限以便清理
test_file.chmod(0o644)
# ---------------------------------------------------------------------- #
# bump_version_alpha
# Version pattern tests
# ---------------------------------------------------------------------- #
class TestBumpVersionAlpha:
"""Test bump_version_alpha function."""
class TestVersionPattern:
"""Test version pattern matching."""
def test_bump_version_alpha_patch(self) -> None:
"""Should bump alpha patch version."""
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
bumpversion.bump_version_alpha("patch")
assert mock_run.called
def test_simple_version(self, tmp_path: Path) -> None:
"""Should match simple version."""
test_file = tmp_path / "test.txt"
test_file.write_text("1.0.0", encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "patch")
assert result == "1.0.1"
def test_version_with_zeros(self, tmp_path: Path) -> None:
"""Should handle versions with zeros correctly."""
test_file = tmp_path / "test.txt"
test_file.write_text("0.0.0", encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "patch")
assert result == "0.0.1"
def test_large_version_numbers(self, tmp_path: Path) -> None:
"""Should handle large version numbers."""
test_file = tmp_path / "test.txt"
test_file.write_text("10.20.30", encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "minor")
assert result == "10.21.0"
def test_version_in_url(self, tmp_path: Path) -> None:
"""Should only match first version in complex text."""
test_file = tmp_path / "test.txt"
test_file.write_text("https://example.com/v1.2.3/download", encoding="utf-8")
result = bumpversion.bump_file_version(test_file, "patch")
assert result == "1.2.4"
# ---------------------------------------------------------------------- #
# TaskSpec definitions
# Edge cases
# ---------------------------------------------------------------------- #
class TestTaskSpecDefinitions:
"""Test that all TaskSpec definitions are valid."""
class TestEdgeCases:
"""Test edge cases and error handling."""
def test_bump_patch_spec(self) -> None:
"""bump_patch spec should be properly defined."""
assert bumpversion.bump_patch.name == "bump_patch"
assert bumpversion.bump_patch.fn is not None
def test_empty_file(self, tmp_path: Path, capsys) -> None:
"""Should handle empty file."""
test_file = tmp_path / "empty.txt"
test_file.write_text("", encoding="utf-8")
def test_bump_minor_spec(self) -> None:
"""bump_minor spec should be properly defined."""
assert bumpversion.bump_minor.name == "bump_minor"
assert bumpversion.bump_minor.fn is not None
result = bumpversion.bump_file_version(test_file, "patch")
def test_bump_major_spec(self) -> None:
"""bump_major spec should be properly defined."""
assert bumpversion.bump_major.name == "bump_major"
assert bumpversion.bump_major.fn is not None
assert result is None
captured = capsys.readouterr()
assert "未找到版本号模式" in captured.out
def test_file_with_special_chars(self, tmp_path: Path) -> None:
"""Should handle file with special characters."""
test_file = tmp_path / "test.txt"
content = "# 中文注释\nversion = 1.0.0\n# 特殊字符: @#$%"
test_file.write_text(content, encoding="utf-8")
# ---------------------------------------------------------------------- #
# main function
# ---------------------------------------------------------------------- #
class TestMain:
"""Test main function."""
result = bumpversion.bump_file_version(test_file, "patch")
def test_main_calls_run_cli(self) -> None:
"""main() should create a CliRunner and call run_cli()."""
with patch.object(px.CliRunner, "run_cli") as mock_run_cli:
bumpversion.main()
assert mock_run_cli.called
assert result == "1.0.1"
updated = test_file.read_text(encoding="utf-8")
assert "# 中文注释" in updated
assert "# 特殊字符: @#$%" in updated
def test_consecutive_bumps(self, tmp_path: Path) -> None:
"""Should handle consecutive version bumps correctly."""
test_file = tmp_path / "test.txt"
test_file.write_text("version = 1.0.0", encoding="utf-8")
# 第一次 bump
result1 = bumpversion.bump_file_version(test_file, "patch")
assert result1 == "1.0.1"
# 第二次 bump
result2 = bumpversion.bump_file_version(test_file, "minor")
assert result2 == "1.1.0"
# 第三次 bump
result3 = bumpversion.bump_file_version(test_file, "major")
assert result3 == "2.0.0"
# 验证最终结果
assert test_file.read_text(encoding="utf-8") == "version = 2.0.0"
uv.lock
Generated
+1 -1
View File
@@ -2184,7 +2184,7 @@ wheels = [
[[package]]
name = "pyflowx"
version = "0.1.12"
version = "0.1.13"
source = { editable = "." }
dependencies = [
{ name = "graphlib-backport", marker = "python_full_version < '3.9'" },