chongming/chongming-utils-task
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1725b6cbdb532431a4c72f1a05722cf5ea533e1a
chongming-utils-task/tests/integration/test_cli.py
T

767 lines
26 KiB
Python
Raw Normal View History

初始提交：chongming-task 任务控制库
2026-10-08 23:47:50 +08:00
"""命令行测试。
CLI 是**另一个进程**，所以这里全部通过 ``--db`` 指向真实的 SQLite 文件，而不是把
``Controller`` 塞进去——那样测的就不是 CLI 了。
重点覆盖两件容易静默出错的事：
* **输出契约**（``--json`` 的形状、退出码），因为脚本要依赖它；
* **错误路径**，尤其是"给一个不存在的任务下指令"——它曾经会成功并留下一条永远不会
被执行的指令。
"""
from __future__ import annotations
import json
import os
from datetime import UTC, datetime
from pathlib import Path
import pytest
from chongming_task.cli.main import (
EXIT_ERROR,
EXIT_OK,
EXIT_TASK_NOT_SUCCESSFUL,
build_parser,
main,
)
from chongming_task.errors import ConfigurationError
from chongming_task.models import TaskSnapshot
from chongming_task.states import TaskStatus
from chongming_task.stores.sqlite import SqliteTaskStore
_TASKS_MODULE = "tasks_demo"
@pytest.fixture
def db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> str:
"""把示例任务模块加进导入路径，并返回一个全新的库路径。"""
import sys
examples = Path(__file__).resolve().parents[2] / "examples"
monkeypatch.syspath_prepend(str(examples))
sys.modules.pop(_TASKS_MODULE, None)
return str(tmp_path / "tasks.db")
def _run_cli(db: str, *argv: str) -> int:
return main(["--db", db, *argv])
# ---------------------------------------------------------------------------
# 参数解析
# ---------------------------------------------------------------------------
def test_parser_requires_a_subcommand() -> None:
with pytest.raises(SystemExit):
build_parser().parse_args([])
def test_parser_help_mentions_every_command() -> None:
text = build_parser().format_help()
for command in (
"tasks",
"run",
"ps",
"status",
"history",
"stats",
"buckets",
"stop",
"pause",
"resume",
):
assert command in text
# ---------------------------------------------------------------------------
# 定义
# ---------------------------------------------------------------------------
def test_tasks_lists_the_registered_definitions(
db: str, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["tasks", "--tasks", _TASKS_MODULE]) == EXIT_OK
out = capsys.readouterr().out
assert "sync" in out
assert "add" in out
assert "两数相加。" in out, "注册项的 docstring 应当作为描述显示"
def test_tasks_on_an_unimportable_module_fails_clearly(
capsys: pytest.CaptureFixture[str],
) -> None:
assert main(["tasks", "--tasks", "no.such.module"]) == EXIT_ERROR
assert "无法导入任务模块" in capsys.readouterr().err
def test_a_module_not_found_error_explains_sys_path(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""模块自身没找到时，必须说清"为什么找不到"。
最常踩的坑：``chongming-task`` 是**入口脚本**，不会把当前目录放进 ``sys.path``——
把模块放在当前目录、然后直接 ``--tasks mymodule`` 会失败，而 "No module named"
本身完全不解释这件事。
"""
monkeypatch.delenv("PYTHONPATH", raising=False)
assert main(["tasks", "--tasks", "definitely_not_installed"]) == EXIT_ERROR
err = capsys.readouterr().err
assert "PYTHONPATH" in err
assert "不会**把当前目录放进 sys.path" in err
assert "python -m chongming_task.cli.main" in err
def test_a_module_found_but_failing_to_import_says_so(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""模块存在、但内部导入失败：不能把矛头指向 sys.path，那是它自己的依赖问题。"""
module = tmp_path / "broken_tasks.py"
module.write_text("import no_such_dependency_here\n", encoding="utf-8")
monkeypatch.syspath_prepend(str(tmp_path))
assert main(["tasks", "--tasks", "broken_tasks"]) == EXIT_ERROR
err = capsys.readouterr().err
assert "找到了，但它导入失败" in err
assert "PYTHONPATH" not in err
def test_the_cli_module_can_be_run_with_python_m() -> None:
"""备选入口：``python -m`` 会把当前目录放进 ``sys.path``，因此根目录的模块也能用。
这条用例钉住"备选入口存在"这个事实——它已经写进了错误提示里，不能悄悄消失。
"""
source = Path(__file__).resolve().parents[2] / "src" / "chongming_task" / "cli" / "main.py"
assert source.is_file()
assert 'if __name__ == "__main__"' in source.read_text(encoding="utf-8")
def test_tasks_on_a_module_without_register_fails_clearly(
capsys: pytest.CaptureFixture[str],
) -> None:
"""约定必须说清楚——否则用户只会看到一句 AttributeError。"""
assert main(["tasks", "--tasks", "json"]) == EXIT_ERROR
assert "register(controller)" in capsys.readouterr().err
# ---------------------------------------------------------------------------
# 提交与执行
# ---------------------------------------------------------------------------
def test_run_can_execute_in_a_child_process(db: str, capsys: pytest.CaptureFixture[str]) -> None:
"""``--executor process``：任务在子进程里跑，快照因此带上 pid。
这是"能被真的杀掉"那条能力在命令行上的入口——前台 ``run`` 遇到不配合的任务时，
另一个终端发 ``stop`` 就能把它连根拔掉（线程只能放弃等待）。
"""
code = _run_cli(
db,
"run",
"--tasks",
_TASKS_MODULE,
"add",
"--arg",
"left=2",
"--arg",
"right=3",
"--executor",
"process",
)
assert code == EXIT_OK
assert "succeeded" in capsys.readouterr().out
store = SqliteTaskStore(db)
try:
snapshot = store.query()[0]
assert snapshot.executor == "process"
assert snapshot.pid is not None
assert snapshot.pid != os.getpid(), "任务确实跑在另一个进程里"
finally:
store.close()
def test_run_executes_and_exits_zero(db: str, capsys: pytest.CaptureFixture[str]) -> None:
code = _run_cli(
db, "run", "--tasks", _TASKS_MODULE, "add", "--arg", "left=2", "--arg", "right=3"
)
assert code == EXIT_OK
out = capsys.readouterr().out
assert "succeeded" in out
store = SqliteTaskStore(db)
try:
snapshots = store.query()
assert len(snapshots) == 1
assert snapshots[0].outcome == {"value": 5}
finally:
store.close()
def test_run_parses_json_typed_arguments(db: str) -> None:
"""``--arg n=3`` 应当得到整数 3，而不是字符串 "3"。"""
assert (
_run_cli(
db, "run", "--tasks", _TASKS_MODULE, "add", "--arg", "left=20", "--arg", "right=22"
)
== EXIT_OK
)
store = SqliteTaskStore(db)
try:
assert store.query()[0].outcome == {"value": 42}
finally:
store.close()
def test_run_on_a_failing_task_exits_three(db: str) -> None:
"""命令成功、任务失败——两者用不同退出码区分，脚本才好判分支。"""
code = _run_cli(db, "run", "--tasks", _TASKS_MODULE, "flaky", "--arg", "fail_times=9")
assert code == EXIT_TASK_NOT_SUCCESSFUL
def test_run_rejects_a_malformed_arg(db: str, capsys: pytest.CaptureFixture[str]) -> None:
code = _run_cli(db, "run", "--tasks", _TASKS_MODULE, "add", "--arg", "noequals")
assert code == EXIT_ERROR
assert "K=V" in capsys.readouterr().err
def test_run_records_the_bucket(db: str) -> None:
code = _run_cli(
db,
"run",
"--tasks",
_TASKS_MODULE,
"add",
"--arg",
"left=1",
"--arg",
"right=1",
"--bucket-key",
"tenant-9",
)
assert code == EXIT_OK
store = SqliteTaskStore(db)
try:
snapshot = store.query()[0]
assert snapshot.bucket_key == "tenant-9"
assert snapshot.bucket is not None
finally:
store.close()
# ---------------------------------------------------------------------------
# 查询
# ---------------------------------------------------------------------------
def test_ps_defaults_to_unfinished_and_all_shows_everything(
db: str, capsys: pytest.CaptureFixture[str]
) -> None:
assert (
_run_cli(db, "run", "--tasks", _TASKS_MODULE, "add", "--arg", "left=1", "--arg", "right=1")
== EXIT_OK
)
capsys.readouterr()
assert _run_cli(db, "ps") == EXIT_OK
assert "没有匹配的任务" in capsys.readouterr().out, "默认只显示未落终态的"
assert _run_cli(db, "ps", "--all") == EXIT_OK
assert "succeeded" in capsys.readouterr().out
def test_status_shows_detail_and_history(db: str, capsys: pytest.CaptureFixture[str]) -> None:
_run_cli(db, "run", "--tasks", _TASKS_MODULE, "add", "--arg", "left=1", "--arg", "right=1")
capsys.readouterr()
assert _run_cli(db, "status", "x") == EXIT_ERROR # 不存在
store = SqliteTaskStore(db)
try:
task_id = store.query()[0].task_id
finally:
store.close()
assert _run_cli(db, "status", task_id) == EXIT_OK
out = capsys.readouterr().out
assert "status" in out
assert "迁移历史" in out
assert "executor:completed" in out
def test_stats_and_buckets(db: str, capsys: pytest.CaptureFixture[str]) -> None:
_run_cli(db, "run", "--tasks", _TASKS_MODULE, "add", "--arg", "left=1", "--arg", "right=1")
capsys.readouterr()
assert _run_cli(db, "stats") == EXIT_OK
out = capsys.readouterr().out
assert "succeeded" in out
assert "total" in out
assert _run_cli(db, "buckets") == EXIT_OK
assert "BUCKET" in capsys.readouterr().out
def test_ps_rejects_an_unknown_status(db: str, capsys: pytest.CaptureFixture[str]) -> None:
assert _run_cli(db, "ps", "--status", "no-such-status") == EXIT_ERROR
assert "未知状态" in capsys.readouterr().err
def test_ps_filters_by_pipeline_run(db: str, capsys: pytest.CaptureFixture[str]) -> None:
"""按管道查看是看板最常问的问题，过滤下推到存储层。"""
from datetime import UTC, datetime
from chongming_task.models import TaskSnapshot
store = SqliteTaskStore(db)
try:
for task_id, run_id in (("a1", "run-a"), ("a2", "run-a"), ("b1", "run-b")):
store.create(
TaskSnapshot(
task_id=task_id,
name="节点",
status=TaskStatus.SUCCEEDED,
status_version=1,
attempt=1,
max_attempts=1,
created_at=datetime(2026, 1, 1, tzinfo=UTC),
ended_at=datetime(2026, 1, 1, 1, tzinfo=UTC),
pipeline_run_id=run_id,
)
)
finally:
store.close()
assert _run_cli(db, "ps", "--all", "--pipeline", "run-a") == EXIT_OK
out = capsys.readouterr().out
assert "a1" in out
assert "a2" in out
assert "b1" not in out
assert _run_cli(db, "ps", "--all", "--pipeline", "no-such-run") == EXIT_OK
assert "没有匹配的任务" in capsys.readouterr().out
# ---------------------------------------------------------------------------
# 输出契约
# ---------------------------------------------------------------------------
def test_json_output_is_primitive_only(db: str, capsys: pytest.CaptureFixture[str]) -> None:
"""``--json`` 的输出必须全是原语，脚本才不必了解 datetime / 枚举的序列化规则。"""
_run_cli(db, "run", "--tasks", _TASKS_MODULE, "add", "--arg", "left=2", "--arg", "right=3")
capsys.readouterr()
assert main(["--db", db, "--json", "ps", "--all"]) == EXIT_OK
payload = json.loads(capsys.readouterr().out)
assert isinstance(payload, list)
entry = payload[0]
assert entry["status"] == "succeeded"
assert entry["outcome"] == '{"value": 5}'
assert all(
value is None or isinstance(value, (str, int, float, bool)) for value in entry.values()
), "出现了非原语字段"
def test_json_error_path_uses_stderr_not_stdout(capsys: pytest.CaptureFixture[str]) -> None:
assert main(["--json", "ps"]) == EXIT_ERROR
captured = capsys.readouterr()
assert captured.out == ""
assert "未指定任务存储" in captured.err
# ---------------------------------------------------------------------------
# 错误路径
# ---------------------------------------------------------------------------
def test_missing_db_is_reported_clearly(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.delenv("CHONGMING_TASK_DB", raising=False)
assert main(["ps"]) == EXIT_ERROR
assert "CHONGMING_TASK_DB" in capsys.readouterr().err
def test_db_can_come_from_the_environment(db: str, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("CHONGMING_TASK_DB", db)
_run_cli(db, "run", "--tasks", _TASKS_MODULE, "add", "--arg", "left=1", "--arg", "right=1")
assert main(["ps", "--all"]) == EXIT_OK
def test_stop_on_an_unknown_task_fails_instead_of_writing_a_dead_command(
db: str, capsys: pytest.CaptureFixture[str]
) -> None:
"""曾经这里的 bug：给不存在的任务下指令会成功，留下一条永远不会被执行的指令，
而调用方还以为"停止已下发"。
"""
_run_cli(db, "run", "--tasks", _TASKS_MODULE, "add", "--arg", "left=1", "--arg", "right=1")
capsys.readouterr()
assert _run_cli(db, "stop", "typo-task-id") == EXIT_ERROR
assert "没有找到任务" in capsys.readouterr().err
store = SqliteTaskStore(db)
try:
assert store.commands() == [], "不该留下任何指令"
finally:
store.close()
def test_control_on_an_existing_task_is_accepted(
db: str, capsys: pytest.CaptureFixture[str]
) -> None:
_run_cli(db, "run", "--tasks", _TASKS_MODULE, "add", "--arg", "left=1", "--arg", "right=1")
capsys.readouterr()
store = SqliteTaskStore(db)
try:
task_id = store.query()[0].task_id
finally:
store.close()
# 任务已落终态，但"找到并受理"仍然成立。
assert _run_cli(db, "stop", task_id, "--reason", "收尾") == EXIT_OK
assert "已请求停止" in capsys.readouterr().out
def test_store_directory_missing_is_reported_clearly(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""SQLite 只会报 "unable to open database file"，完全看不出是目录不存在。"""
missing = tmp_path / "not-created" / "tasks.db"
assert main(["--db", str(missing), "ps"]) == EXIT_ERROR
assert "目录不存在" in capsys.readouterr().err
# ---------------------------------------------------------------------------
# 重启恢复
# ---------------------------------------------------------------------------
def _plant_zombie(db: str, task_id: str = "zombie") -> None:
"""造一个"上次崩溃留下的"任务行：running 且租约早已过期。"""
from datetime import UTC, datetime, timedelta
from chongming_task.models import TaskSnapshot
store = SqliteTaskStore(db)
try:
old = datetime.now(UTC) - timedelta(minutes=10)
store.create(
TaskSnapshot(
task_id=task_id,
name="遗留任务",
status=TaskStatus.RUNNING,
status_version=3,
attempt=1,
max_attempts=1,
created_at=old,
started_at=old,
lease_owner="worker-old",
lease_until=datetime.now(UTC) - timedelta(minutes=9),
)
)
finally:
store.close()
def test_reconcile_dry_run_writes_nothing(db: str, capsys: pytest.CaptureFixture[str]) -> None:
_plant_zombie(db)
assert _run_cli(db, "reconcile", "--dry-run") == EXIT_OK
out = capsys.readouterr().out
assert "marked-lost" in out
assert "预演" in out
store = SqliteTaskStore(db)
try:
snapshot = store.get("zombie")
assert snapshot is not None
assert snapshot.status is TaskStatus.RUNNING, "预演不该写状态"
assert store.history("zombie") == []
finally:
store.close()
def test_reconcile_marks_zombies(db: str, capsys: pytest.CaptureFixture[str]) -> None:
_plant_zombie(db)
assert _run_cli(db, "reconcile", "--reason", "worker-old 崩溃") == EXIT_OK
assert "已把 1 个僵尸任务标记为 lost" in capsys.readouterr().out
store = SqliteTaskStore(db)
try:
snapshot = store.get("zombie")
assert snapshot is not None
assert snapshot.status is TaskStatus.LOST
assert store.history("zombie")[0].reason == "worker-old 崩溃"
finally:
store.close()
def test_reconcile_with_nothing_to_do(db: str, capsys: pytest.CaptureFixture[str]) -> None:
assert _run_cli(db, "reconcile") == EXIT_OK
assert "没有需要处理的任务" in capsys.readouterr().out
def test_reconcile_json_report(db: str, capsys: pytest.CaptureFixture[str]) -> None:
_plant_zombie(db)
assert main(["--db", db, "--json", "reconcile", "--dry-run"]) == EXIT_OK
payload = json.loads(capsys.readouterr().out)
assert payload["dry_run"] is True
assert payload["entries"][0]["task_id"] == "zombie"
assert payload["entries"][0]["previous_status"] == "running"
assert payload["entries"][0]["outcome"] == "marked-lost"
# ---------------------------------------------------------------------------
# 节点与桶归属（自治节点 + 统一平面）
# ---------------------------------------------------------------------------
def test_bucket_spec_supports_ranges_and_lists() -> None:
from chongming_task.cli.main import _parse_bucket_spec
assert _parse_bucket_spec("0-3") == frozenset({0, 1, 2, 3})
assert _parse_bucket_spec("0-1,5,7-8") == frozenset({0, 1, 5, 7, 8})
assert _parse_bucket_spec(" 2 , 4 ") == frozenset({2, 4})
assert _parse_bucket_spec("3-3") == frozenset({3})
def test_bucket_spec_rejects_bad_input() -> None:
from chongming_task.cli.main import _parse_bucket_spec
for bad in ("3-1", "a", "0-", "1,x"):
with pytest.raises(ConfigurationError):
_parse_bucket_spec(bad)
with pytest.raises(ConfigurationError, match="没有解析出任何桶"):
_parse_bucket_spec(" , ")
def test_ps_mine_filters_to_owned_buckets(db: str, capsys: pytest.CaptureFixture[str]) -> None:
"""自治架构里最常看的视图："我这一份现在怎样"。"""
_plant_zombie(db, "zombie") # 无桶任务
store = SqliteTaskStore(db)
try:
for task_id, bucket in (("in-a", 1), ("in-b", 40)):
store.create(
TaskSnapshot(
task_id=task_id,
name="节点任务",
bucket=bucket,
bucket_key=f"key-{bucket}",
status=TaskStatus.SUCCEEDED,
status_version=1,
attempt=1,
max_attempts=1,
created_at=datetime(2026, 1, 1, tzinfo=UTC),
ended_at=datetime(2026, 1, 1, 1, tzinfo=UTC),
)
)
finally:
store.close()
assert main(["--db", db, "--owned-buckets", "0-31", "ps", "--all", "--mine"]) == EXIT_OK
out = capsys.readouterr().out
assert "in-a" in out
assert "in-b" not in out, "别把别的桶的任务算进来"
assert "zombie" not in out, "无桶任务不属于任何限定了归属的节点"
def test_ps_mine_requires_owned_buckets(db: str, capsys: pytest.CaptureFixture[str]) -> None:
assert main(["--db", db, "ps", "--mine"]) == EXIT_ERROR
assert "--owned-buckets" in capsys.readouterr().err
def test_reconcile_is_scoped_to_owned_buckets(db: str, capsys: pytest.CaptureFixture[str]) -> None:
"""节点只清理自己桶里的僵尸——这是自治架构的核心不变量。"""
store = SqliteTaskStore(db)
try:
for task_id, bucket in (("mine", 1), ("theirs", 40)):
old = datetime(2026, 1, 1, tzinfo=UTC)
store.create(
TaskSnapshot(
task_id=task_id,
name="遗留任务",
bucket=bucket,
bucket_key=f"key-{bucket}",
status=TaskStatus.RUNNING,
status_version=3,
attempt=1,
max_attempts=1,
created_at=old,
started_at=old,
lease_owner="dead",
lease_until=old,
)
)
finally:
store.close()
assert main(["--db", db, "--owned-buckets", "0-31", "reconcile"]) == EXIT_OK
out = capsys.readouterr().out
assert "mine" in out
assert "theirs" not in out
store = SqliteTaskStore(db)
try:
assert store.get("mine").status is TaskStatus.LOST # type: ignore[union-attr]
assert store.get("theirs").status is TaskStatus.RUNNING, ( # type: ignore[union-attr]
"动了别的节点的任务"
)
finally:
store.close()
def test_buckets_view_marks_owned_buckets(db: str, capsys: pytest.CaptureFixture[str]) -> None:
assert main(["--db", db, "--num-buckets", "4", "--owned-buckets", "0-1", "buckets"]) == EXIT_OK
out = capsys.readouterr().out
marked = [line for line in out.splitlines() if line.startswith("*")]
assert len(marked) == 2, f"应当恰好标出两个桶，实际：{marked}"
def test_an_out_of_range_owned_bucket_is_reported(
db: str, capsys: pytest.CaptureFixture[str]
) -> None:
"""桶号越界必须在启动时报错——否则症状是"任务永远排队且没有任何报错"。"""
assert main(["--db", db, "--owned-buckets", "0-999", "ps"]) == EXIT_ERROR
assert "桶号越界" in capsys.readouterr().err
def test_run_respects_owned_buckets(db: str, capsys: pytest.CaptureFixture[str]) -> None:
"""``run`` 必须**真的**受归属约束。
曾经 ``run`` 自己构造 ``Controller(...)``，把 ``--owned-buckets`` 静默忽略了——
参数被接受却不起作用，比报错更难查。
"""
from chongming_task import BucketPolicy
policy = BucketPolicy(num_buckets=64)
mine = next(f"tenant-{i}" for i in range(999) if policy.bucket_for(f"tenant-{i}") < 16)
theirs = next(f"tenant-{i}" for i in range(999) if policy.bucket_for(f"tenant-{i}") >= 48)
# 限定了归属却没给 bucket_key → 必须报错（否则会产出没人认领的任务）
assert (
main(
[
"--db",
db,
"--num-buckets",
"64",
"--owned-buckets",
"0-15",
"run",
"--tasks",
_TASKS_MODULE,
"add",
"--arg",
"left=1",
"--arg",
"right=1",
"--task-id",
"r1",
]
)
== EXIT_ERROR
)
assert "必须给出 bucket_key" in capsys.readouterr().err
# 落在别人的桶 → 必须报错
assert (
main(
[
"--db",
db,
"--num-buckets",
"64",
"--owned-buckets",
"0-15",
"run",
"--tasks",
_TASKS_MODULE,
"add",
"--arg",
"left=1",
"--arg",
"right=1",
"--task-id",
"r2",
"--bucket-key",
theirs,
]
)
== EXIT_ERROR
)
assert "不属于本节点" in capsys.readouterr().err
# 落在自己的桶 → 正常执行
assert (
main(
[
"--db",
db,
"--num-buckets",
"64",
"--owned-buckets",
"0-15",
"run",
"--tasks",
_TASKS_MODULE,
"add",
"--arg",
"left=20",
"--arg",
"right=22",
"--task-id",
"r3",
"--bucket-key",
mine,
]
)
== EXIT_OK
)
assert "succeeded" in capsys.readouterr().out
def test_a_module_with_only_task_decorators_works(
db: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""只用 ``@task`` 装饰、**没有** ``register()`` 的模块也必须能用。
``@task`` 是后来加的写法；CLI 若只认 ``register()``，那个 API 在命令行下就废了。
"""
module = tmp_path / "decorated_only.py"
module.write_text(
"from chongming_task import task\n"
"\n"
"@task(tags={'io'})\n"
"def doubled(ctx, value: int) -> int:\n"
" '''把数字翻倍。'''\n"
" del ctx\n"
" return value * 2\n",
encoding="utf-8",
)
monkeypatch.syspath_prepend(str(tmp_path))
assert main(["tasks", "--tasks", "decorated_only"]) == EXIT_OK
out = capsys.readouterr().out
assert "doubled" in out
assert "把数字翻倍。" in out
assert (
main(["--db", db, "run", "--tasks", "decorated_only", "doubled", "--arg", "value=21"])
== EXIT_OK
)
assert "succeeded" in capsys.readouterr().out
def test_a_module_with_neither_register_nor_task_is_reported(
capsys: pytest.CaptureFixture[str],
) -> None:
assert main(["tasks", "--tasks", "json"]) == EXIT_ERROR
assert "既没有 register" in capsys.readouterr().err
Reference in New Issue Copy Permalink