Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def deduping_treats_different_calls_to_same_task_differently(self):
body = Mock()
t1 = Task(body)
pre = [call(t1, 5), call(t1, 7), call(t1, 5)]
t2 = Task(Mock(), pre=pre)
c = Collection(t1=t1, t2=t2)
e = Executor(collection=c)
e.execute("t2")
# Does not call the second t1(5)
param_list = []
for body_call in body.call_args_list:
assert isinstance(body_call[0][0], Context)
param_list.append(body_call[0][1])
assert set(param_list) == {5, 7}
def equality_testing(self):
t1 = Task(_func, name="foo")
t2 = Task(_func, name="foo")
assert t1 == t2
t3 = Task(_func, name="bar")
assert t1 != t3
def _call_objs(self):
# Setup
pre_body, post_body = Mock(), Mock()
t1 = Task(pre_body)
t2 = Task(post_body)
t3 = Task(
Mock(),
pre=[call(t1, 5, foo="bar")],
post=[call(t2, 7, biz="baz")],
)
c = Collection(t1=t1, t2=t2, t3=t3)
e = Executor(collection=c)
e.execute("t3")
# Pre-task asserts
args, kwargs = pre_body.call_args
assert kwargs == {"foo": "bar"}
assert isinstance(args[0], Context)
assert args[1] == 5
# Post-task asserts
args, kwargs = post_body.call_args
assert kwargs == {"biz": "baz"}
assert isinstance(args[0], Context)
def bundled_namespace_help_includes_subcommand_listing(self):
t1, t2 = Task(Mock()), Task(Mock())
coll = Collection(task1=t1, task2=t2)
p = Program(namespace=coll)
# Spot checks for expected bits, so we don't have to change
# this every time core args change.
for expected in (
# Usage line changes somewhat
"Usage: myapp [--core-opts] [--subcommand-opts] ...\n", # noqa
# Core options are still present
"Core options:\n",
"--echo",
# Subcommands are listed
"Subcommands:\n",
" task1",
" task2",
):
stdout, _ = run("myapp --help", program=p, invoke=False)
def returns_Task_instances_by_default(self):
@task
def mytask(c):
pass
assert isinstance(mytask, Task)
def klass_kwarg_allows_overriding_class_used(self):
class MyTask(Task):
pass
@task(klass=MyTask)
def mytask(c):
pass
assert isinstance(mytask, MyTask)
def setup(self):
self.task1 = Task(Mock(return_value=7))
self.task2 = Task(Mock(return_value=10), pre=[self.task1])
self.task3 = Task(Mock(), pre=[self.task1])
self.task4 = Task(Mock(return_value=15), post=[self.task1])
self.contextualized = Task(Mock())
coll = Collection()
coll.add_task(self.task1, name="task1")
coll.add_task(self.task2, name="task2")
coll.add_task(self.task3, name="task3")
coll.add_task(self.task4, name="task4")
coll.add_task(self.contextualized, name="contextualized")
self.executor = Executor(collection=coll)
def has_useful_repr(self):
i = repr(Task(_func))
assert "_func" in i, "'func' not found in {!r}".format(i)
e = repr(Task(_func, name="funky"))
assert "funky" in e, "'funky' not found in {!r}".format(e)
assert "_func" not in e, "'_func' unexpectedly seen in {!r}".format(e)
# Init collection; invoke picks this up as the main "list of tasks"
ns = Collection()
# Automatically collect tasks from submodules
for fname in os.listdir(THIS_DIR):
# Does this look like a module that we want?
if fname.startswith("_") or not fname.endswith(".py"):
continue
modname = fname[:-3]
# Import it
m = __import__(modname, level=1, fromlist=[], globals=globals())
# Collect all tasks and collections
collections, tasks = {}, {}
for name in dir(m):
ob = getattr(m, name)
if isinstance(ob, Task):
tasks[name] = ob
elif isinstance(ob, Collection):
collections[name] = ob
# Add collections
for name, ob in collections.items():
ns.add_collection(ob, name)
# Add tasks that are not already in a collection
for name, ob in tasks.items():
add_task = True
for c in collections.values():
if ob in c.tasks.values():
add_task = False
if add_task:
ns.add_task(ob, name)
def delete_foreign_tasks(locals):
here = locals['__name__']
for (name, func) in locals.items():
if isinstance(func, Task) and func.__module__ != here and 'tasks.' in func.__module__:
locals.pop(name)