diff --git a/.trae/rules/git-commit-message.md b/.trae/rules/git-commit-message.md new file mode 100644 index 0000000..3eba4d4 --- /dev/null +++ b/.trae/rules/git-commit-message.md @@ -0,0 +1,11 @@ +--- +alwaysApply: true +scene: git_message +--- + +在此处编写规则,自定义 AI 生成提交信息的风格。 + +## 提交信息格式 +- 提交信息必须使用中文。 +- 提交信息必须包含变更的类型(例如 "fix"、"feat"、"refactor" 等)。 +- 提交信息必须尽简洁明了,不要超过一段落。 diff --git a/src/pyflowx/tasks/system.py b/src/pyflowx/tasks/system.py index eb60dc4..73ea061 100644 --- a/src/pyflowx/tasks/system.py +++ b/src/pyflowx/tasks/system.py @@ -113,10 +113,7 @@ def write_file(path: str, content: str, encoding: str = "utf-8") -> px.TaskSpec: """写入文件任务.""" def write(): - try: - with open(path, "w", encoding=encoding) as f: - f.write(content) - except Exception as e: - print(f"写入文件 {path} 失败: {e}") + p = Path(path) + p.write_text(content, encoding=encoding) return px.TaskSpec(f"write_file_{path}", fn=write, verbose=True) diff --git a/tests/test_conditions.py b/tests/test_conditions.py index e8a371b..2e467f4 100644 --- a/tests/test_conditions.py +++ b/tests/test_conditions.py @@ -301,3 +301,59 @@ def test_dir_exists_false(tmp_path: Path): missing = tmp_path / "nonexistent" cond = BuiltinConditions.DIR_EXISTS(missing) assert cond({}) is False + + +def test_builtin_is_windows_returns_module_condition(): + """BuiltinConditions.IS_WINDOWS() 应返回模块级 IS_WINDOWS.""" + assert BuiltinConditions.IS_WINDOWS() is IS_WINDOWS + + +def test_builtin_is_linux_returns_module_condition(): + """BuiltinConditions.IS_LINUX() 应返回模块级 IS_LINUX.""" + assert BuiltinConditions.IS_LINUX() is IS_LINUX + + +def test_builtin_is_macos_returns_module_condition(): + """BuiltinConditions.IS_MACOS() 应返回模块级 IS_MACOS.""" + assert BuiltinConditions.IS_MACOS() is IS_MACOS + + +def test_builtin_is_posix_returns_module_condition(): + """BuiltinConditions.IS_POSIX() 应返回模块级 IS_POSIX.""" + assert BuiltinConditions.IS_POSIX() is IS_POSIX + + +def test_file_content_exists_missing_file(tmp_path: Path): + """FILE_CONTENT_EXISTS 文件不存在时返回 False.""" + cond = BuiltinConditions.FILE_CONTENT_EXISTS(tmp_path / "missing.txt", "x") + assert cond({}) is False + + +def test_file_content_exists_contains_content(tmp_path: Path): + """FILE_CONTENT_EXISTS 文件包含内容时返回 True.""" + f = tmp_path / "f.txt" + f.write_text("hello world", encoding="utf-8") + cond = BuiltinConditions.FILE_CONTENT_EXISTS(f, "world") + assert cond({}) is True + + +def test_file_content_exists_not_contains_content(tmp_path: Path): + """FILE_CONTENT_EXISTS 文件不包含内容时返回 False.""" + f = tmp_path / "f.txt" + f.write_text("hello", encoding="utf-8") + cond = BuiltinConditions.FILE_CONTENT_EXISTS(f, "missing") + assert cond({}) is False + + +def test_file_content_exists_decode_error_returns_false(tmp_path: Path): + """FILE_CONTENT_EXISTS 读取非 UTF-8 文件应返回 False(解码异常被吞).""" + f = tmp_path / "bin.dat" + f.write_bytes(b"\xff\xfe\x00bad") + cond = BuiltinConditions.FILE_CONTENT_EXISTS(f, "x") + assert cond({}) is False + + +def test_dep_matches_missing_dep_returns_false(): + """DEP_MATCHES 依赖不存在时应返回 False(覆盖 if not in ctx 分支).""" + cond = BuiltinConditions.DEP_MATCHES("missing", lambda _v: True) + assert cond({}) is False diff --git a/tests/test_graph.py b/tests/test_graph.py index 17f2d58..8d2b0ac 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -319,6 +319,79 @@ def test_compose_function() -> None: assert "a1" in resolved["cmd_b"] +def test_graph_composer_expand_refs_multiple_refs_chain() -> None: + """expand_refs 多个 ref 应串联依赖:后一个 ref 首任务依赖前一个 ref 末任务.""" + graph_a = px.Graph.from_specs([px.TaskSpec("a1", _fn)]) + graph_c = px.Graph.from_specs([px.TaskSpec("c1", _fn)]) + graph_b = px.Graph.from_specs([px.TaskSpec("b1", _fn)]) + graph_b._pending_refs = ["cmd_a", "cmd_c"] + + composer = GraphComposer({"cmd_a": graph_a, "cmd_c": graph_c, "cmd_b": graph_b}) + resolved = composer.resolve_all() + + # c1 应依赖 a1(后 ref 首任务依赖前 ref 末任务) + assert "a1" in resolved["cmd_b"] + assert "c1" in resolved["cmd_b"] + assert "b1" in resolved["cmd_b"] + c1_spec = resolved["cmd_b"].all_specs()["c1"] + assert "a1" in c1_spec.depends_on + + +def test_graph_composer_expand_refs_ref_returns_empty() -> None: + """expand_refs 引用空图时,previous_ref_last_task 保持 None,original_specs 走 else 分支.""" + graph_empty = px.Graph.from_specs([]) + graph_b = px.Graph.from_specs([px.TaskSpec("b1", _fn)]) + graph_b._pending_refs = ["empty_cmd"] + + composer = GraphComposer({"empty_cmd": graph_empty, "cmd_b": graph_b}) + resolved = composer.resolve_all() + + # b1 保留,无额外依赖 + assert "b1" in resolved["cmd_b"] + b1_spec = resolved["cmd_b"].all_specs()["b1"] + assert b1_spec.depends_on == () + + +def test_graph_composer_expand_refs_multiple_original_specs_serialized() -> None: + """expand_refs 多个 original_specs 应串行依赖,且首个依赖 ref 末任务.""" + graph_a = px.Graph.from_specs([px.TaskSpec("a1", _fn)]) + graph_b = px.Graph.from_specs([ + px.TaskSpec("b1", _fn), + px.TaskSpec("b2", _fn), + px.TaskSpec("b3", _fn), + ]) + graph_b._pending_refs = ["cmd_a"] + + composer = GraphComposer({"cmd_a": graph_a, "cmd_b": graph_b}) + resolved = composer.resolve_all() + + specs = resolved["cmd_b"].all_specs() + # b1 依赖 a1(ref 末任务) + assert "a1" in specs["b1"].depends_on + # b2 依赖 b1,b3 依赖 b2(串行) + assert "b1" in specs["b2"].depends_on + assert "b2" in specs["b3"].depends_on + + +def test_graph_composer_parse_ref_dot_notation_success() -> None: + """parse_ref 'cmd.task' 形式应返回对应单个 TaskSpec.""" + graph_a = px.Graph.from_specs([px.TaskSpec("a1", _fn), px.TaskSpec("a2", _fn)]) + composer = GraphComposer({"cmd_a": graph_a}) + + result = composer.parse_ref("cmd_a.a2", "cmd_b") + assert len(result) == 1 + assert result[0].name == "a2" + + +def test_graph_composer_parse_ref_dot_notation_cmd_not_found() -> None: + """parse_ref 'missing.task' 形式应检测命令不存在.""" + graph_a = px.Graph.from_specs([px.TaskSpec("a1", _fn)]) + composer = GraphComposer({"cmd_a": graph_a}) + + with pytest.raises(ValueError, match="引用的命令 'missing' 不存在"): + _ = composer.parse_ref("missing.task", "cmd_b") + + # ---------------------------------------------------------------------- # # resolved_spec defaults 测试 # ---------------------------------------------------------------------- # diff --git a/tests/test_system.py b/tests/test_system.py index be530a1..229f1b5 100644 --- a/tests/test_system.py +++ b/tests/test_system.py @@ -2,11 +2,12 @@ import os import subprocess +from pathlib import Path import pytest from pyflowx.conditions import Constants -from pyflowx.tasks.system import clr, reset_icon_cache, setenv, which +from pyflowx.tasks.system import clr, reset_icon_cache, setenv, setenv_group, which, write_file def test_clr_creates_task_spec() -> None: @@ -189,3 +190,57 @@ def test_which_not_found(monkeypatch: pytest.MonkeyPatch, capsys: pytest.Capture spec.fn() captured = capsys.readouterr() assert "nonexistent_cmd -> 未找到" in captured.out + + +def test_write_file_creates_task_spec() -> None: + """write_file() 应创建带 verbose 的 TaskSpec。""" + spec = write_file("/tmp/unused", "x") + assert spec.name == "write_file_/tmp/unused" + assert spec.verbose is True + + +def test_write_file_writes_content(tmp_path: Path) -> None: + """write_file() 应将内容写入指定文件.""" + f = tmp_path / "out.txt" + spec = write_file(str(f), "hello world") + assert spec.fn is not None + spec.fn() + assert f.read_text(encoding="utf-8") == "hello world" + + +def test_write_file_with_encoding(tmp_path: Path) -> None: + """write_file() 应支持指定编码.""" + f = tmp_path / "out.txt" + spec = write_file(str(f), "中文", encoding="utf-8") + assert spec.fn is not None + spec.fn() + assert f.read_text(encoding="utf-8") == "中文" + + +def test_write_file_failure_propagates(tmp_path: Path) -> None: + """write_file() 写入失败应抛出异常(不吞异常).""" + # 父目录不存在时写入应抛 FileNotFoundError + missing = tmp_path / "no_such_dir" / "out.txt" + spec = write_file(str(missing), "x") + assert spec.fn is not None + with pytest.raises(FileNotFoundError): + spec.fn() + + +def test_setenv_group_creates_specs() -> None: + """setenv_group() 应为每个环境变量创建 TaskSpec.""" + envs = {"VAR_A": "1", "VAR_B": "2"} + specs = setenv_group(envs) + assert len(specs) == 2 + assert specs[0].name == "setenv_var_a" + assert specs[1].name == "setenv_var_b" + + +def test_setenv_group_default_mode(monkeypatch: pytest.MonkeyPatch) -> None: + """setenv_group(default=True) 不应覆盖已存在的环境变量.""" + monkeypatch.setenv("PYFLOWX_GROUP_EXISTS", "original") + specs = setenv_group({"PYFLOWX_GROUP_EXISTS": "new"}, default=True) + for spec in specs: + assert spec.fn is not None + spec.fn() + assert os.environ["PYFLOWX_GROUP_EXISTS"] == "original"