swh_scheduler = <swh.scheduler.backend.SchedulerBackend object at 0x7fe19ca8bdd8>
def test_schedule_register_lister(swh_scheduler):
result = invoke(
swh_scheduler,
[
"register-lister",
"test",
"arg1",
"arg2",
> "key=value",
],
)
.tox/py3/lib/python3.7/site-packages/swh/scheduler/tests/test_cli_add_forge_now.py:111:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/py3/lib/python3.7/site-packages/swh/scheduler/tests/test_cli_add_forge_now.py:18: in invoke
scheduler, args=["add-forge-now", *args], catch_exceptions=catch_exceptions
.tox/py3/lib/python3.7/site-packages/swh/scheduler/tests/test_cli.py:45: in invoke
raise result.exception
.tox/py3/lib/python3.7/site-packages/click/testing.py:408: in invoke
return_value = cli.main(args=args or (), prog_name=prog_name, **extra)
.tox/py3/lib/python3.7/site-packages/click/core.py:1055: in main
rv = self.invoke(ctx)
.tox/py3/lib/python3.7/site-packages/click/core.py:1657: in invoke
return _process_result(sub_ctx.command.invoke(sub_ctx))
.tox/py3/lib/python3.7/site-packages/click/core.py:1657: in invoke
return _process_result(sub_ctx.command.invoke(sub_ctx))
.tox/py3/lib/python3.7/site-packages/click/core.py:1404: in invoke
return ctx.invoke(self.callback, **ctx.params)
.tox/py3/lib/python3.7/site-packages/click/core.py:760: in invoke
return __callback(*args, **kwargs)
.tox/py3/lib/python3.7/site-packages/click/decorators.py:26: in new_func
return f(get_current_context(), *args, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
ctx = <click.core.Context object at 0x7fe19c9cefd0>, lister_name = 'test'
options = ('arg1', 'arg2', 'key=value')
@add_forge_now.command("register-lister")
@click.argument("lister_name", nargs=1, required=True)
@click.argument("options", nargs=-1)
@click.pass_context
def register_lister_cli(
ctx,
lister_name,
options,
):
"""Register the lister tasks in the scheduler.
Depending on the environment, we'll want different policies:
- staging: only the "full" but limited listing (as "oneshot" task) of disabled origins
- production: both "full" and "incremental" (if that exists) listing (as "recurring"
task). The "full" will be triggered asap, the "incremental" will be triggered the
next day.
"""
from .utils import parse_options, task_add
scheduler = ctx.obj["scheduler"]
environment = ctx.obj["environment"]
# Map the associated task types for the lister
task_type_names = {
lister_type: f"lister-{lister_name}-" for lister_type in ["full", "incremental"]
}
task_types = {}
for type_name, task_type_name in task_type_names.items():
task_type = scheduler.get_task_type(task_type_name)
if task_type:
task_types[type_name] = task_type
if not task_types:
> raise ValueError(f"Unknown lister type {lister_name}.")
E ValueError: Unknown lister type test.
.tox/py3/lib/python3.7/site-packages/swh/scheduler/cli/add_forge_now.py:73: ValueError
TEST RESULT
TEST RESULT
- Run At
- Dec 8 2022, 12:47 PM