Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| ab8faec863 | |||
| 936a009212 | |||
| f10f8d09a6 | |||
| 0d6a78f320 | |||
| c9a4192c85 | |||
| 0afdb54e5c |
+23
-23
@@ -17,31 +17,31 @@ license = { text = "MIT" }
|
||||
name = "pyflowx"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.8"
|
||||
version = "0.1.10"
|
||||
version = "0.2.0"
|
||||
|
||||
[project.scripts]
|
||||
autofmt = "pyflowx.cli.autofmt:main"
|
||||
bumpver = "pyflowx.cli.bumpversion:main"
|
||||
clr = "pyflowx.cli.clearscreen:main"
|
||||
emlman = "pyflowx.cli.emlmanager:main"
|
||||
envpy = "pyflowx.cli.envpy:main"
|
||||
envqt = "pyflowx.cli.envqt:main"
|
||||
envrs = "pyflowx.cli.envrs:main"
|
||||
filedate = "pyflowx.cli.filedate:main"
|
||||
filelvl = "pyflowx.cli.filelevel:main"
|
||||
foldback = "pyflowx.cli.folderback:main"
|
||||
foldzip = "pyflowx.cli.folderzip:main"
|
||||
gitt = "pyflowx.cli.gittool:main"
|
||||
hfdown = "pyflowx.cli.hfdownload:main"
|
||||
lscalc = "pyflowx.cli.lscalc:main"
|
||||
packtool = "pyflowx.cli.packtool:main"
|
||||
pdftool = "pyflowx.cli.pdftool:main"
|
||||
piptool = "pyflowx.cli.piptool:main"
|
||||
pymake = "pyflowx.cli.pymake:main"
|
||||
scrcap = "pyflowx.cli.screenshot:main"
|
||||
sshcopy = "pyflowx.cli.sshcopyid:main"
|
||||
taskk = "pyflowx.cli.taskkill:main"
|
||||
wch = "pyflowx.cli.which:main"
|
||||
autofmt = "pyflowx.cli.autofmt:main"
|
||||
bumpversion = "pyflowx.cli.bumpversion:main"
|
||||
clr = "pyflowx.cli.clearscreen:main"
|
||||
emlman = "pyflowx.cli.emlmanager:main"
|
||||
envpy = "pyflowx.cli.envpy:main"
|
||||
envqt = "pyflowx.cli.envqt:main"
|
||||
envrs = "pyflowx.cli.envrs:main"
|
||||
filedate = "pyflowx.cli.filedate:main"
|
||||
filelvl = "pyflowx.cli.filelevel:main"
|
||||
foldback = "pyflowx.cli.folderback:main"
|
||||
foldzip = "pyflowx.cli.folderzip:main"
|
||||
gitt = "pyflowx.cli.gittool:main"
|
||||
hfdown = "pyflowx.cli.hfdownload:main"
|
||||
lscalc = "pyflowx.cli.lscalc:main"
|
||||
packtool = "pyflowx.cli.packtool:main"
|
||||
pdftool = "pyflowx.cli.pdftool:main"
|
||||
piptool = "pyflowx.cli.piptool:main"
|
||||
pymake = "pyflowx.cli.pymake:main"
|
||||
scrcap = "pyflowx.cli.screenshot:main"
|
||||
sshcopy = "pyflowx.cli.sshcopyid:main"
|
||||
taskk = "pyflowx.cli.taskkill:main"
|
||||
wch = "pyflowx.cli.which:main"
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = [
|
||||
|
||||
@@ -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.10"
|
||||
__version__ = "0.2.0"
|
||||
|
||||
__all__ = [
|
||||
"IS_LINUX",
|
||||
|
||||
+232
-72
@@ -5,97 +5,257 @@
|
||||
|
||||
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"]
|
||||
|
||||
# 针对不同文件类型的版本号匹配模式
|
||||
# pyproject.toml: version = "X.Y.Z" 或 version = 'X.Y.Z'
|
||||
_PYPROJECT_VERSION_PATTERN = re.compile(
|
||||
r'(?:^|\n)\s*version\s*=\s*["\']'
|
||||
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-]+)*))?"
|
||||
r'["\']',
|
||||
re.MULTILINE,
|
||||
)
|
||||
|
||||
# __init__.py: __version__ = "X.Y.Z" 或 __version__ = 'X.Y.Z'
|
||||
_INIT_VERSION_PATTERN = re.compile(
|
||||
r'(?:^|\n)\s*__version__\s*=\s*["\']'
|
||||
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-]+)*))?"
|
||||
r'["\']',
|
||||
re.MULTILINE,
|
||||
)
|
||||
|
||||
|
||||
def bump_version(part: str = "patch", tag: bool = False, commit: bool = False) -> None:
|
||||
"""递增版本号.
|
||||
def _get_pattern_for_file(file_name: str) -> re.Pattern[str] | None:
|
||||
"""根据文件类型获取对应的正则表达式.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
part : str
|
||||
file_name : str
|
||||
文件名
|
||||
|
||||
Returns
|
||||
-------
|
||||
re.Pattern[str] | None
|
||||
对应的正则表达式,如果无法确定则返回 None
|
||||
"""
|
||||
if file_name == "pyproject.toml":
|
||||
return _PYPROJECT_VERSION_PATTERN
|
||||
if file_name == "__init__.py":
|
||||
return _INIT_VERSION_PATTERN
|
||||
return None
|
||||
|
||||
|
||||
def _calculate_new_version(major: int, minor: int, patch: int, part: BumpVersionType) -> str:
|
||||
"""计算新版本号.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
major : int
|
||||
当前主版本号
|
||||
minor : int
|
||||
当前次版本号
|
||||
patch : int
|
||||
当前补丁版本号
|
||||
part : BumpVersionType
|
||||
要更新的部分
|
||||
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
新版本号
|
||||
"""
|
||||
if part == "major":
|
||||
return f"{major + 1}.0.0"
|
||||
if part == "minor":
|
||||
return f"{major}.{minor + 1}.0"
|
||||
return f"{major}.{minor}.{patch + 1}"
|
||||
|
||||
|
||||
def _build_replacement_string(original_match: str, new_version: str, file_name: str) -> str:
|
||||
"""构建替换字符串,保留原始格式.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
original_match : str
|
||||
原始匹配的字符串
|
||||
new_version : str
|
||||
新版本号
|
||||
file_name : str
|
||||
文件名
|
||||
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
替换字符串
|
||||
"""
|
||||
quote_char = '"' if '"' in original_match else "'"
|
||||
|
||||
if file_name == "pyproject.toml":
|
||||
prefix_match = re.match(r'(\s*version\s*=\s*)["\']', original_match)
|
||||
prefix = prefix_match.group(1) if prefix_match else "version = "
|
||||
return f"{prefix}{quote_char}{new_version}{quote_char}"
|
||||
|
||||
if file_name == "__init__.py":
|
||||
prefix_match = re.match(r'(\s*__version__\s*=\s*)["\']', original_match)
|
||||
prefix = prefix_match.group(1) if prefix_match else "__version__ = "
|
||||
return f"{prefix}{quote_char}{new_version}{quote_char}"
|
||||
|
||||
return new_version
|
||||
|
||||
|
||||
def bump_file_version(file_path: Path, part: BumpVersionType = "patch") -> str | None:
|
||||
"""更新文件中的版本号.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
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
|
||||
|
||||
# 获取文件对应的正则表达式
|
||||
pattern = _get_pattern_for_file(file_path.name)
|
||||
|
||||
# 对于未知文件类型,尝试两种模式
|
||||
if pattern:
|
||||
match = pattern.search(content)
|
||||
else:
|
||||
match = _PYPROJECT_VERSION_PATTERN.search(content) or _INIT_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"))
|
||||
|
||||
# 计算新版本号
|
||||
new_version = _calculate_new_version(major, minor, patch, part)
|
||||
|
||||
# 构建替换字符串
|
||||
original_match = match.group(0)
|
||||
replacement = _build_replacement_string(original_match, new_version, file_path.name)
|
||||
|
||||
# 更新文件内容
|
||||
content = content.replace(original_match, replacement)
|
||||
|
||||
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
|
||||
|
||||
|
||||
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}")
|
||||
|
||||
@@ -23,6 +23,7 @@ EXCLUDE_DIRS = [
|
||||
".tox",
|
||||
".pytest_cache",
|
||||
"node_modules",
|
||||
".ruff_cache",
|
||||
]
|
||||
EXCLUDE_CMDS = [arg for d in EXCLUDE_DIRS for arg in ["-e", d]]
|
||||
|
||||
|
||||
@@ -49,6 +49,7 @@ test_coverage: px.TaskSpec = px.TaskSpec(
|
||||
ruff_lint: px.TaskSpec = px.TaskSpec("lint", cmd=["ruff", "check", "--fix", "--unsafe-fixes"])
|
||||
ruff_format: px.TaskSpec = px.TaskSpec("format", cmd=["ruff", "format", "."], depends_on=("lint",))
|
||||
typecheck: px.TaskSpec = px.TaskSpec("pyrefly_check", cmd=["pyrefly", "check", "."])
|
||||
git_add_all: px.TaskSpec = px.TaskSpec("git_add_all", cmd=["git", "add", "-A"])
|
||||
bump: px.TaskSpec = px.TaskSpec("bumpversion", cmd=["bumpversion", "-t"])
|
||||
doc: px.TaskSpec = px.TaskSpec("doc", cmd=["sphinx-build", "-b", "html", "docs", "docs/_build"])
|
||||
git_push: px.TaskSpec = px.TaskSpec("git_push", cmd=["git", "push"])
|
||||
@@ -86,7 +87,10 @@ def main():
|
||||
📦 发布命令:
|
||||
pymake pb - 发布到 PyPI (twine + hatch)
|
||||
|
||||
💡 常用工作流:
|
||||
� 版本管理:
|
||||
pymake bump - 自动升级版本号并提交修改 (清理 + 检查 + 格式化 + git add + bumpversion)
|
||||
|
||||
�💡 常用工作流:
|
||||
1. 日常开发: pymake lint && pymake t
|
||||
2. 构建发布包: pymake ba
|
||||
3. 多版本兼容性测试: pymake tox
|
||||
@@ -113,7 +117,8 @@ def main():
|
||||
# 清理命令
|
||||
"c": px.Graph.from_specs([git_clean]),
|
||||
# 开发工具
|
||||
"bump": px.Graph.from_specs(["c", "tc", bump]),
|
||||
"bump": px.Graph.from_specs(["c", "tc", git_add_all, bump]),
|
||||
"bumpmi": px.Graph.from_specs([px.TaskSpec("bumpversion_minor", cmd=["bumpversion", "minor"])]),
|
||||
"cov": px.Graph.from_specs([git_clean, test_coverage]),
|
||||
"doc": px.Graph.from_specs([doc]),
|
||||
"lint": px.Graph.from_specs([ruff_lint, ruff_format]),
|
||||
|
||||
+22
-14
@@ -163,15 +163,13 @@ class CliRunner:
|
||||
if not pending_refs:
|
||||
return graph
|
||||
|
||||
# 收集所有TaskSpec(包括原始图中的)
|
||||
# 收集所有TaskSpec(按正确顺序:先引用,后原始TaskSpec)
|
||||
all_specs: list[TaskSpec[Any]] = []
|
||||
for spec in graph.all_specs().values():
|
||||
all_specs.append(spec)
|
||||
|
||||
# 记录每个引用展开后的所有任务名,用于建立依赖链
|
||||
previous_ref_last_task: str | None = None
|
||||
|
||||
# 解析每个引用,并建立依赖关系
|
||||
# 先解析每个引用,并建立依赖关系
|
||||
for ref in pending_refs:
|
||||
expanded_specs = self._parse_ref(ref, current_cmd)
|
||||
|
||||
@@ -190,17 +188,27 @@ class CliRunner:
|
||||
|
||||
all_specs.extend(expanded_specs)
|
||||
|
||||
# 如果原始图中有TaskSpec,让它们依赖于最后一个引用的任务
|
||||
# 然后添加原始图中的TaskSpec,并让它们按顺序执行
|
||||
original_specs = list(graph.all_specs().values())
|
||||
if previous_ref_last_task and original_specs:
|
||||
# 为每个原始TaskSpec添加依赖
|
||||
for i, original_task in enumerate(original_specs):
|
||||
# 只为第一个原始任务添加依赖,或者为没有依赖的任务添加依赖
|
||||
if i == 0 or not original_task.depends_on:
|
||||
updated_task = replace(
|
||||
original_task, depends_on=tuple({*original_task.depends_on, previous_ref_last_task})
|
||||
)
|
||||
all_specs[all_specs.index(original_task)] = updated_task
|
||||
if original_specs:
|
||||
# 第一个原始TaskSpec依赖于最后一个引用的任务
|
||||
if previous_ref_last_task:
|
||||
first_original = original_specs[0]
|
||||
updated_first = replace(
|
||||
first_original, depends_on=tuple({*first_original.depends_on, previous_ref_last_task})
|
||||
)
|
||||
all_specs.append(updated_first)
|
||||
else:
|
||||
# 如果没有引用,直接添加第一个原始TaskSpec
|
||||
all_specs.append(original_specs[0])
|
||||
|
||||
# 后续的原始TaskSpec依赖于前一个原始TaskSpec
|
||||
for i in range(1, len(original_specs)):
|
||||
current_task = original_specs[i]
|
||||
previous_task_name = original_specs[i - 1].name
|
||||
# 更新依赖,确保顺序执行
|
||||
updated_task = replace(current_task, depends_on=tuple({*current_task.depends_on, previous_task_name}))
|
||||
all_specs.append(updated_task)
|
||||
|
||||
# 创建新的图(不包含引用)
|
||||
return Graph.from_specs(all_specs)
|
||||
|
||||
+229
-73
@@ -2,105 +2,261 @@
|
||||
|
||||
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 / "pyproject.toml"
|
||||
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 / "pyproject.toml"
|
||||
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 / "pyproject.toml"
|
||||
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 / "pyproject.toml"
|
||||
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 / "pyproject.toml"
|
||||
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 / "__init__.py"
|
||||
test_file.write_text('__version__ = "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") == '__version__ = "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 project version, not dependencies."""
|
||||
test_file = tmp_path / "pyproject.toml"
|
||||
content = """
|
||||
[project]
|
||||
version = "1.0.0"
|
||||
dependencies = ["lib >= 2.0.0", "other >= 3.0.0"]
|
||||
"""
|
||||
test_file.write_text(content, encoding="utf-8")
|
||||
|
||||
result = bumpversion.bump_file_version(test_file, "patch")
|
||||
|
||||
assert result == "1.0.1"
|
||||
updated = test_file.read_text(encoding="utf-8")
|
||||
assert 'version = "1.0.1"' in updated
|
||||
# 确保 dependencies 中的版本没有被更新
|
||||
assert "lib >= 2.0.0" in updated
|
||||
assert "other >= 3.0.0" in updated
|
||||
|
||||
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.toml"
|
||||
test_file.write_text('version = "1.0.0"', encoding="utf-8")
|
||||
# 设置为只读
|
||||
test_file.chmod(0o444)
|
||||
|
||||
try:
|
||||
with pytest.raises(Exception): # noqa: B017
|
||||
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 / "__init__.py"
|
||||
test_file.write_text('__version__ = "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 / "__init__.py"
|
||||
test_file.write_text('__version__ = "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 / "__init__.py"
|
||||
test_file.write_text('__version__ = "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 not match version in URL or other contexts."""
|
||||
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")
|
||||
|
||||
# 不应该匹配 URL 中的版本号
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------- #
|
||||
# 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 / "__init__.py"
|
||||
content = '# 中文注释\n__version__ = "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 / "__init__.py"
|
||||
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"'
|
||||
|
||||
@@ -194,3 +194,306 @@ class TestCommandReferences:
|
||||
|
||||
# Verify total layers
|
||||
assert len(layers) == 4
|
||||
|
||||
def test_execution_order_multiple_original_tasks(self) -> None:
|
||||
"""Should execute multiple original TaskSpecs in correct order."""
|
||||
task1 = px.TaskSpec("task1", cmd=["echo", "1"])
|
||||
task2 = px.TaskSpec("task2", cmd=["echo", "2"])
|
||||
task3 = px.TaskSpec("task3", cmd=["echo", "3"])
|
||||
task4 = px.TaskSpec("task4", cmd=["echo", "4"])
|
||||
task5 = px.TaskSpec("task5", cmd=["echo", "5"])
|
||||
|
||||
runner = px.CliRunner(
|
||||
strategy="sequential",
|
||||
graphs={
|
||||
"cmd1": px.Graph.from_specs([task1]),
|
||||
"cmd2": px.Graph.from_specs([task2]),
|
||||
"all": px.Graph.from_specs(["cmd1", "cmd2", task3, task4, task5]),
|
||||
},
|
||||
)
|
||||
|
||||
# Check execution order through layers
|
||||
layers = runner.graphs["all"].layers()
|
||||
|
||||
# Layer 1: task1 (cmd1)
|
||||
assert "task1" in layers[0]
|
||||
|
||||
# Layer 2: task2 (cmd2)
|
||||
assert "task2" in layers[1]
|
||||
|
||||
# Layer 3: task3 (first original TaskSpec)
|
||||
assert "task3" in layers[2]
|
||||
|
||||
# Layer 4: task4 (second original TaskSpec)
|
||||
assert "task4" in layers[3]
|
||||
|
||||
# Layer 5: task5 (third original TaskSpec)
|
||||
assert "task5" in layers[4]
|
||||
|
||||
# Verify total layers
|
||||
assert len(layers) == 5
|
||||
|
||||
def test_execution_order_with_internal_dependencies(self) -> None:
|
||||
"""Should preserve internal dependencies within referenced commands."""
|
||||
task1 = px.TaskSpec("task1", cmd=["echo", "1"])
|
||||
task2 = px.TaskSpec("task2", cmd=["echo", "2"], depends_on=("task1",))
|
||||
task3 = px.TaskSpec("task3", cmd=["echo", "3"])
|
||||
task4 = px.TaskSpec("task4", cmd=["echo", "4"])
|
||||
|
||||
runner = px.CliRunner(
|
||||
strategy="sequential",
|
||||
graphs={
|
||||
"cmd1": px.Graph.from_specs([task1, task2]),
|
||||
"cmd2": px.Graph.from_specs([task3]),
|
||||
"all": px.Graph.from_specs(["cmd1", "cmd2", task4]),
|
||||
},
|
||||
)
|
||||
|
||||
# Check execution order through layers
|
||||
layers = runner.graphs["all"].layers()
|
||||
|
||||
# Layer 1: task1
|
||||
assert "task1" in layers[0]
|
||||
|
||||
# Layer 2: task2 (depends on task1)
|
||||
assert "task2" in layers[1]
|
||||
|
||||
# Layer 3: task3 (cmd2, depends on task2)
|
||||
assert "task3" in layers[2]
|
||||
|
||||
# Layer 4: task4 (original TaskSpec, depends on task3)
|
||||
assert "task4" in layers[3]
|
||||
|
||||
# Verify total layers
|
||||
assert len(layers) == 4
|
||||
|
||||
def test_execution_order_pymake_bump_scenario(self) -> None:
|
||||
"""Should execute pymake bump command in correct order."""
|
||||
# Simulate pymake bump scenario
|
||||
git_clean = px.TaskSpec("git_clean", cmd=["echo", "clean"])
|
||||
typecheck = px.TaskSpec("typecheck", cmd=["echo", "typecheck"])
|
||||
lint = px.TaskSpec("lint", cmd=["echo", "lint"])
|
||||
format_task = px.TaskSpec("format", cmd=["echo", "format"], depends_on=("lint",))
|
||||
git_add_all = px.TaskSpec("git_add_all", cmd=["echo", "git add -A"])
|
||||
bump = px.TaskSpec("bumpversion", cmd=["echo", "bumpversion -t"])
|
||||
|
||||
runner = px.CliRunner(
|
||||
strategy="sequential",
|
||||
graphs={
|
||||
"c": px.Graph.from_specs([git_clean]),
|
||||
"tc": px.Graph.from_specs([typecheck, "lint"]),
|
||||
"lint": px.Graph.from_specs([lint, format_task]),
|
||||
"bump": px.Graph.from_specs(["c", "tc", git_add_all, bump]),
|
||||
},
|
||||
)
|
||||
|
||||
# Check execution order through layers
|
||||
layers = runner.graphs["bump"].layers()
|
||||
|
||||
# Layer 1: git_clean (c)
|
||||
assert "git_clean" in layers[0]
|
||||
|
||||
# Layer 2: lint (tc.lint, depends on git_clean)
|
||||
assert "lint" in layers[1]
|
||||
|
||||
# Layer 3: format (tc.lint.format, depends on lint)
|
||||
assert "format" in layers[2]
|
||||
|
||||
# Layer 4: typecheck (tc.typecheck, depends on format)
|
||||
assert "typecheck" in layers[3]
|
||||
|
||||
# Layer 5: git_add_all (original TaskSpec, depends on typecheck)
|
||||
assert "git_add_all" in layers[4]
|
||||
|
||||
# Layer 6: bumpversion (original TaskSpec, depends on git_add_all)
|
||||
assert "bumpversion" in layers[5]
|
||||
|
||||
# Verify total layers
|
||||
assert len(layers) == 6
|
||||
|
||||
def test_execution_order_only_references(self) -> None:
|
||||
"""Should execute only references without original TaskSpecs."""
|
||||
task1 = px.TaskSpec("task1", cmd=["echo", "1"])
|
||||
task2 = px.TaskSpec("task2", cmd=["echo", "2"])
|
||||
task3 = px.TaskSpec("task3", cmd=["echo", "3"])
|
||||
|
||||
runner = px.CliRunner(
|
||||
strategy="sequential",
|
||||
graphs={
|
||||
"cmd1": px.Graph.from_specs([task1]),
|
||||
"cmd2": px.Graph.from_specs([task2]),
|
||||
"cmd3": px.Graph.from_specs([task3]),
|
||||
"all": px.Graph.from_specs(["cmd1", "cmd2", "cmd3"]),
|
||||
},
|
||||
)
|
||||
|
||||
# Check execution order through layers
|
||||
layers = runner.graphs["all"].layers()
|
||||
|
||||
# Layer 1: task1 (cmd1)
|
||||
assert "task1" in layers[0]
|
||||
|
||||
# Layer 2: task2 (cmd2, depends on task1)
|
||||
assert "task2" in layers[1]
|
||||
|
||||
# Layer 3: task3 (cmd3, depends on task2)
|
||||
assert "task3" in layers[2]
|
||||
|
||||
# Verify total layers
|
||||
assert len(layers) == 3
|
||||
|
||||
def test_execution_order_only_original_tasks(self) -> None:
|
||||
"""Should execute only original TaskSpecs without references."""
|
||||
task1 = px.TaskSpec("task1", cmd=["echo", "1"])
|
||||
task2 = px.TaskSpec("task2", cmd=["echo", "2"])
|
||||
task3 = px.TaskSpec("task3", cmd=["echo", "3"])
|
||||
|
||||
runner = px.CliRunner(
|
||||
strategy="sequential",
|
||||
graphs={
|
||||
"all": px.Graph.from_specs([task1, task2, task3]),
|
||||
},
|
||||
)
|
||||
|
||||
# Check execution order through layers
|
||||
layers = runner.graphs["all"].layers()
|
||||
|
||||
# All tasks should be in layer 1 (no dependencies)
|
||||
assert "task1" in layers[0]
|
||||
assert "task2" in layers[0]
|
||||
assert "task3" in layers[0]
|
||||
|
||||
# Verify total layers
|
||||
assert len(layers) == 1
|
||||
|
||||
def test_execution_order_single_reference(self) -> None:
|
||||
"""Should execute single reference correctly."""
|
||||
task1 = px.TaskSpec("task1", cmd=["echo", "1"])
|
||||
task2 = px.TaskSpec("task2", cmd=["echo", "2"])
|
||||
|
||||
runner = px.CliRunner(
|
||||
strategy="sequential",
|
||||
graphs={
|
||||
"cmd1": px.Graph.from_specs([task1, task2]),
|
||||
"all": px.Graph.from_specs(["cmd1"]),
|
||||
},
|
||||
)
|
||||
|
||||
# Check execution order through layers
|
||||
layers = runner.graphs["all"].layers()
|
||||
|
||||
# Should have the same structure as cmd1
|
||||
assert "task1" in layers[0]
|
||||
assert "task2" in layers[0]
|
||||
|
||||
# Verify total layers
|
||||
assert len(layers) == 1
|
||||
|
||||
def test_execution_order_deep_nesting(self) -> None:
|
||||
"""Should execute deeply nested references correctly."""
|
||||
task1 = px.TaskSpec("task1", cmd=["echo", "1"])
|
||||
task2 = px.TaskSpec("task2", cmd=["echo", "2"])
|
||||
task3 = px.TaskSpec("task3", cmd=["echo", "3"])
|
||||
task4 = px.TaskSpec("task4", cmd=["echo", "4"])
|
||||
task5 = px.TaskSpec("task5", cmd=["echo", "5"])
|
||||
|
||||
runner = px.CliRunner(
|
||||
strategy="sequential",
|
||||
graphs={
|
||||
"cmd1": px.Graph.from_specs([task1]),
|
||||
"cmd2": px.Graph.from_specs(["cmd1", task2]),
|
||||
"cmd3": px.Graph.from_specs(["cmd2", task3]),
|
||||
"cmd4": px.Graph.from_specs(["cmd3", task4]),
|
||||
"cmd5": px.Graph.from_specs(["cmd4", task5]),
|
||||
},
|
||||
)
|
||||
|
||||
# Check execution order through layers
|
||||
layers = runner.graphs["cmd5"].layers()
|
||||
|
||||
# Should execute in order: task1 -> task2 -> task3 -> task4 -> task5
|
||||
assert "task1" in layers[0]
|
||||
assert "task2" in layers[1]
|
||||
assert "task3" in layers[2]
|
||||
assert "task4" in layers[3]
|
||||
assert "task5" in layers[4]
|
||||
|
||||
# Verify total layers
|
||||
assert len(layers) == 5
|
||||
|
||||
def test_execution_order_with_parallel_tasks_in_reference(self) -> None:
|
||||
"""Should handle parallel tasks within referenced commands."""
|
||||
task1 = px.TaskSpec("task1", cmd=["echo", "1"])
|
||||
task2 = px.TaskSpec("task2", cmd=["echo", "2"])
|
||||
task3 = px.TaskSpec("task3", cmd=["echo", "3"])
|
||||
task4 = px.TaskSpec("task4", cmd=["echo", "4"])
|
||||
|
||||
runner = px.CliRunner(
|
||||
strategy="sequential",
|
||||
graphs={
|
||||
"cmd1": px.Graph.from_specs([task1, task2]), # Parallel tasks
|
||||
"cmd2": px.Graph.from_specs([task3, task4]), # Parallel tasks
|
||||
"all": px.Graph.from_specs(["cmd1", "cmd2"]),
|
||||
},
|
||||
)
|
||||
|
||||
# Check execution order through layers
|
||||
layers = runner.graphs["all"].layers()
|
||||
|
||||
# Layer 1: task1 and task2 (cmd1, parallel)
|
||||
assert "task1" in layers[0]
|
||||
assert "task2" in layers[0]
|
||||
|
||||
# Layer 2: task3 and task4 (cmd2, depends on cmd1's last task)
|
||||
# Note: Both task3 and task4 should depend on the last task of cmd1
|
||||
assert "task3" in layers[1]
|
||||
assert "task4" in layers[1]
|
||||
|
||||
# Verify total layers
|
||||
assert len(layers) == 2
|
||||
|
||||
def test_execution_order_complex_mixed_scenario(self) -> None:
|
||||
"""Should handle complex mixed scenario with references and TaskSpecs."""
|
||||
# Create a complex scenario
|
||||
clean = px.TaskSpec("clean", cmd=["echo", "clean"])
|
||||
build1 = px.TaskSpec("build1", cmd=["echo", "build1"])
|
||||
build2 = px.TaskSpec("build2", cmd=["echo", "build2"], depends_on=("build1",))
|
||||
test1 = px.TaskSpec("test1", cmd=["echo", "test1"])
|
||||
test2 = px.TaskSpec("test2", cmd=["echo", "test2"])
|
||||
package = px.TaskSpec("package", cmd=["echo", "package"])
|
||||
deploy = px.TaskSpec("deploy", cmd=["echo", "deploy"])
|
||||
|
||||
runner = px.CliRunner(
|
||||
strategy="sequential",
|
||||
graphs={
|
||||
"clean": px.Graph.from_specs([clean]),
|
||||
"build": px.Graph.from_specs([build1, build2]),
|
||||
"test": px.Graph.from_specs([test1, test2]),
|
||||
"release": px.Graph.from_specs(["clean", "build", "test", package, deploy]),
|
||||
},
|
||||
)
|
||||
|
||||
# Check execution order through layers
|
||||
layers = runner.graphs["release"].layers()
|
||||
|
||||
# Layer 1: clean
|
||||
assert "clean" in layers[0]
|
||||
|
||||
# Layer 2: build1 (depends on clean)
|
||||
assert "build1" in layers[1]
|
||||
|
||||
# Layer 3: build2 (depends on build1)
|
||||
assert "build2" in layers[2]
|
||||
|
||||
# Layer 4: test1 and test2 (depends on build2)
|
||||
assert "test1" in layers[3]
|
||||
assert "test2" in layers[3]
|
||||
|
||||
# Layer 5: package (depends on test1/test2)
|
||||
assert "package" in layers[4]
|
||||
|
||||
# Layer 6: deploy (depends on package)
|
||||
assert "deploy" in layers[5]
|
||||
|
||||
# Verify total layers
|
||||
assert len(layers) == 6
|
||||
|
||||
Reference in New Issue
Block a user