84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
|
"""公开 API 的表面测试。
|
|||
|
|
|
|||
|
|
``chongming_task`` 的 ``__init__`` 是**嵌入其他应用**的入口，因此它本身的正确性
|
|||
|
|
值得一条用例：漏导出某个名字、或者导出顺序导致循环导入，都会让集成方第一步就失败。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import chongming_task
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_version_is_exposed() -> None:
|
|||
|
|
assert chongming_task.__version__
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_all_has_no_duplicates() -> None:
|
|||
|
|
"""重复导出会让"到底哪个生效"变得不确定。
|
|||
|
|
|
|||
|
|
顺序**不在这里**断言：``__all__`` 的排序由 ruff 的 ``RUF022``（isort 风格，
|
|||
|
|
全大写常量优先）负责，那是 lint 的职责。写一条与 linter 规则不一致的断言，
|
|||
|
|
只会让两边互相打架。
|
|||
|
|
"""
|
|||
|
|
names = chongming_task.__all__
|
|||
|
|
assert len(names) == len(set(names))
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_every_exported_name_resolves() -> None:
|
|||
|
|
"""``__all__`` 里写了却取不到，等于给集成方埋了一个 ImportError。"""
|
|||
|
|
missing = [name for name in chongming_task.__all__ if not hasattr(chongming_task, name)]
|
|||
|
|
assert missing == []
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_primary_integration_surface_is_importable() -> None:
|
|||
|
|
"""这一段就是 README 里"嵌入你自己的应用"的最小示例。"""
|
|||
|
|
from chongming_task import ( # noqa: F401 - 只为验证可导入
|
|||
|
|
Controller,
|
|||
|
|
InMemoryTaskStore,
|
|||
|
|
SqliteTaskStore,
|
|||
|
|
TaskStatus,
|
|||
|
|
TaskStore,
|
|||
|
|
run_task,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
controller = Controller(max_parallel=2)
|
|||
|
|
assert controller.registry.names() == ()
|
|||
|
|
assert TaskStatus.PENDING.value == "pending"
|
|||
|
|
assert callable(run_task)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_store_implementations_satisfy_the_protocol() -> None:
|
|||
|
|
"""存储实现必须满足契约——第三方后端也照这条自证。"""
|
|||
|
|
from chongming_task import InMemoryTaskStore, SqliteTaskStore, TaskStore
|
|||
|
|
|
|||
|
|
assert isinstance(InMemoryTaskStore(), TaskStore)
|
|||
|
|
|
|||
|
|
store = SqliteTaskStore(":memory:")
|
|||
|
|
try:
|
|||
|
|
assert isinstance(store, TaskStore)
|
|||
|
|
finally:
|
|||
|
|
# 不关掉连接会在解释器退出时触发 ResourceWarning；本项目的 pytest 配置
|
|||
|
|
# 把 warning 升级为 error，因此泄漏是会被抓住的。
|
|||
|
|
store.close()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_controller_accepts_a_store() -> None:
|
|||
|
|
"""存储已经接线。
|
|||
|
|
|
|||
|
|
这条用例原先钉的是"尚未接线"——当 ``Controller(store=...)`` 真的开始工作时，
|
|||
|
|
它会自己失败并提示删掉。现在它钉的是接线本身。
|
|||
|
|
"""
|
|||
|
|
from chongming_task import Controller, InMemoryTaskStore
|
|||
|
|
|
|||
|
|
store = InMemoryTaskStore()
|
|||
|
|
controller = Controller(store=store, max_parallel=2)
|
|||
|
|
assert controller.store is store
|
|||
|
|
assert controller.persist_failures == ()
|
|||
|
|
assert controller.owner
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_private_helpers_are_not_exported() -> None:
|
|||
|
|
"""以下划线开头的实现细节不应出现在公开表面上。"""
|
|||
|
|
leaked = [name for name in chongming_task.__all__ if name.startswith("_")]
|
|||
|
|
assert leaked == []
|