zhou
12d9f2f647
CI / Lint, Typecheck & Test (push) Successful in 1m56s
将 gitt a/i 命令改用 fn job 包装(git_add_commit/git_init_add_commit), 内部检查 has_files() 和 not_has_git_repo() 条件,避免无更改时 git commit 报错。修正 has_files() 实现为检查 git status --porcelain 而非目录文件。
254 lines
7.3 KiB
Python
254 lines
7.3 KiB
Python
"""``pyflowx.registry`` 模块的单元测试."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
import pyflowx as px
|
|
from pyflowx.registry import _REGISTRY, FnRegistry, get_fn, has_fn, register_fn
|
|
|
|
|
|
@pytest.fixture
|
|
def clear_registry() -> Any:
|
|
"""每个测试前后清空 registry, 但保留 _ops 模块自动注册的函数.
|
|
|
|
_ops 模块在首次导入时注册函数, 模块不会重新执行, 因此清空后无法重新注册.
|
|
保存原始状态并在 teardown 时恢复, 确保 TestOpsModules 等不使用此 fixture 的
|
|
测试仍能看到 _ops 的注册.
|
|
"""
|
|
saved = dict(_REGISTRY)
|
|
_REGISTRY.clear()
|
|
yield
|
|
_REGISTRY.clear()
|
|
_REGISTRY.update(saved)
|
|
|
|
|
|
class TestRegisterFn:
|
|
"""``register_fn`` 装饰器测试."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear(self, clear_registry: Any) -> None:
|
|
"""每个测试前后清空 registry."""
|
|
|
|
def test_register_with_explicit_name(self) -> None:
|
|
"""使用显式名称注册函数."""
|
|
|
|
@register_fn("custom_name")
|
|
def _func(x: int) -> int:
|
|
return x * 2
|
|
|
|
assert has_fn("custom_name")
|
|
assert get_fn("custom_name")(5) == 10
|
|
|
|
def test_register_without_parentheses(self) -> None:
|
|
"""无括号使用 ``@register_fn`` 时以 ``__name__`` 注册."""
|
|
|
|
@register_fn
|
|
def my_func(x: int) -> int:
|
|
return x + 1
|
|
|
|
assert has_fn("my_func")
|
|
assert get_fn("my_func")(10) == 11
|
|
|
|
def test_register_with_none_uses_func_name(self) -> None:
|
|
"""``@register_fn(None)`` 等价于使用 ``__name__``."""
|
|
|
|
@register_fn(None)
|
|
def auto_named(x: int) -> int:
|
|
return x
|
|
|
|
assert has_fn("auto_named")
|
|
|
|
def test_register_preserves_function(self) -> None:
|
|
"""装饰后函数仍可直接调用."""
|
|
|
|
@register_fn("calc")
|
|
def _calc(a: int, b: int) -> int:
|
|
return a + b
|
|
|
|
assert _calc(2, 3) == 5
|
|
|
|
def test_duplicate_registration_raises(self) -> None:
|
|
"""重复注册同名函数抛出 ValueError."""
|
|
with pytest.raises(ValueError, match="已注册"):
|
|
|
|
@register_fn("dup")
|
|
def _first() -> None:
|
|
pass
|
|
|
|
@register_fn("dup")
|
|
def _second() -> None:
|
|
pass
|
|
|
|
def test_duplicate_registration_without_parentheses(self) -> None:
|
|
"""无括号模式下重复注册也抛出 ValueError."""
|
|
with pytest.raises(ValueError, match="已注册"):
|
|
|
|
@register_fn
|
|
def shared_name() -> None:
|
|
pass
|
|
|
|
@register_fn
|
|
def shared_name() -> None: # noqa: F811
|
|
pass
|
|
|
|
|
|
class TestGetFn:
|
|
"""``get_fn`` 函数测试."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear(self, clear_registry: Any) -> None:
|
|
"""每个测试前后清空 registry."""
|
|
|
|
def test_get_registered_function(self) -> None:
|
|
"""获取已注册的函数."""
|
|
|
|
@register_fn("target")
|
|
def _target(x: int) -> int:
|
|
return x * 3
|
|
|
|
retrieved = get_fn("target")
|
|
assert retrieved is _target
|
|
|
|
def test_get_unregistered_raises_keyerror(self) -> None:
|
|
"""获取未注册的函数抛出 KeyError."""
|
|
with pytest.raises(KeyError, match="not_found"):
|
|
get_fn("not_found")
|
|
|
|
|
|
class TestHasFn:
|
|
"""``has_fn`` 函数测试."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear(self, clear_registry: Any) -> None:
|
|
"""每个测试前后清空 registry."""
|
|
|
|
def test_has_registered(self) -> None:
|
|
"""已注册返回 True."""
|
|
|
|
@register_fn("exists")
|
|
def _exists() -> None:
|
|
pass
|
|
|
|
assert has_fn("exists") is True
|
|
|
|
def test_has_not_registered(self) -> None:
|
|
"""未注册返回 False."""
|
|
assert has_fn("nope") is False
|
|
|
|
|
|
class TestFnRegistry:
|
|
"""``FnRegistry`` 类测试."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear(self, clear_registry: Any) -> None:
|
|
"""每个测试前后清空 registry."""
|
|
|
|
def test_register_method(self) -> None:
|
|
"""``FnRegistry.register`` 等价于 ``register_fn``."""
|
|
|
|
@FnRegistry.register("via_class")
|
|
def _func(x: int) -> int:
|
|
return x
|
|
|
|
assert FnRegistry.has("via_class")
|
|
assert FnRegistry.get("via_class")(7) == 7
|
|
|
|
def test_register_method_without_name(self) -> None:
|
|
"""``FnRegistry.register(None)`` 使用函数名."""
|
|
|
|
@FnRegistry.register()
|
|
def class_auto(x: int) -> int:
|
|
return x
|
|
|
|
assert FnRegistry.has("class_auto")
|
|
|
|
def test_get_method(self) -> None:
|
|
"""``FnRegistry.get`` 获取已注册函数."""
|
|
|
|
@register_fn("klass")
|
|
def _klass() -> str:
|
|
return "hello"
|
|
|
|
assert FnRegistry.get("klass")() == "hello"
|
|
|
|
def test_has_method_false(self) -> None:
|
|
"""``FnRegistry.has`` 未注册返回 False."""
|
|
assert FnRegistry.has("missing") is False
|
|
|
|
def test_clear_method(self) -> None:
|
|
"""``FnRegistry.clear`` 清空注册表."""
|
|
|
|
@register_fn("temp")
|
|
def _temp() -> None:
|
|
pass
|
|
|
|
assert FnRegistry.has("temp")
|
|
FnRegistry.clear()
|
|
assert not FnRegistry.has("temp")
|
|
|
|
def test_names_method(self) -> None:
|
|
"""``FnRegistry.names`` 返回所有注册名."""
|
|
|
|
@register_fn("alpha")
|
|
def _alpha() -> None:
|
|
pass
|
|
|
|
@register_fn("beta")
|
|
def _beta() -> None:
|
|
pass
|
|
|
|
names = FnRegistry.names()
|
|
assert "alpha" in names
|
|
assert "beta" in names
|
|
|
|
def test_names_returns_copy(self) -> None:
|
|
"""``names`` 返回列表副本, 修改不影响内部状态."""
|
|
|
|
@register_fn("orig")
|
|
def _orig() -> None:
|
|
pass
|
|
|
|
names = FnRegistry.names()
|
|
names.append("fake")
|
|
assert not FnRegistry.has("fake")
|
|
|
|
|
|
class TestOpsModules:
|
|
"""``_ops`` 子模块导入后函数自动注册的集成测试.
|
|
|
|
这些测试不使用 ``_clear_registry`` fixture, 因为 ``_ops`` 模块在首次导入时
|
|
注册函数, 清空后将无法重新注册 (模块不会再次执行).
|
|
"""
|
|
|
|
def test_all_ops_functions_registered(self) -> None:
|
|
"""导入 ``_ops`` 后所有函数应已注册."""
|
|
import inspect
|
|
|
|
from pyflowx.cli._ops import dev, files, media, system
|
|
|
|
for module in (files, dev, media, system):
|
|
for name in module.__all__:
|
|
obj = getattr(module, name)
|
|
if not inspect.isfunction(obj):
|
|
continue
|
|
assert px.has_fn(name), f"{module.__name__}.{name} 未注册"
|
|
|
|
def test_total_function_count(self) -> None:
|
|
"""注册函数总数 = 18+15+18+11 = 62."""
|
|
from pyflowx.cli._ops import dev, files, media, system # noqa: F401
|
|
|
|
all_names = px.FnRegistry.names()
|
|
assert len(all_names) == 62
|
|
|
|
def test_specific_functions_callable(self) -> None:
|
|
"""关键注册函数可调用."""
|
|
from pyflowx.cli._ops import dev, files, media, system
|
|
|
|
assert px.get_fn("get_file_timestamp") is files.get_file_timestamp
|
|
assert px.get_fn("bump_file_version") is dev.bump_file_version
|
|
assert px.get_fn("pdf_merge") is media.pdf_merge
|
|
assert px.get_fn("ssh_copy_id") is system.ssh_copy_id
|