1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
import contextlib
from typing import Tuple
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from celery.bin.celery import celery
@pytest.fixture(autouse=True)
def reset_command_params_between_each_test():
with contextlib.ExitStack() as stack:
for command in celery.commands.values():
# We only need shallow copy -- preload options are appended to the list,
# existing options are kept as-is
params_copy = command.params[:]
patch_instance = patch.object(command, "params", params_copy)
stack.enter_context(patch_instance)
yield
@pytest.mark.parametrize(
"subcommand_with_params",
[
("purge", "-f"),
("shell",),
]
)
def test_preload_options(subcommand_with_params: Tuple[str, ...], isolated_cli_runner: CliRunner):
# Verify commands like shell and purge can accept preload options.
# Projects like Pyramid-Celery's ini option should be valid preload
# options.
res_without_preload = isolated_cli_runner.invoke(
celery,
["-A", "t.unit.bin.proj.app", *subcommand_with_params, "--ini", "some_ini.ini"],
catch_exceptions=False,
)
assert "No such option: --ini" in res_without_preload.output
assert res_without_preload.exit_code == 2
res_with_preload = isolated_cli_runner.invoke(
celery,
[
"-A",
"t.unit.bin.proj.pyramid_celery_app",
*subcommand_with_params,
"--ini",
"some_ini.ini",
],
catch_exceptions=False,
)
assert res_with_preload.exit_code == 0, res_with_preload.output
|
