Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def percolates_to_subcollection_names(self):
@task
def my_task(c):
pass
coll = Collection(inner_coll=Collection(my_task))
contexts = coll.to_contexts()
assert contexts[0].name == "inner-coll.my-task"
def access_merges_from_subcollections(self):
inner = Collection("inner", self.task)
inner.configure({"foo": "bar"})
self.root.configure({"biz": "baz"})
# With no inner collection
assert set(self.root.configuration().keys()) == {"biz"}
# With inner collection
self.root.add_collection(inner)
keys = set(self.root.configuration("inner.task").keys())
assert keys == {"foo", "biz"}
def accepts_load_path_kwarg(self):
assert Collection().loaded_from is None
assert Collection(loaded_from="a/path").loaded_from == "a/path"
def _nested_underscores(self, auto_dash_names=None):
@task(aliases=["other_name"])
def my_task(c):
pass
@task(aliases=["other_inner"])
def inner_task(c):
pass
# NOTE: explicitly not giving kwarg to subcollection; this
# tests that the top-level namespace performs the inverse
# transformation when necessary.
sub = Collection("inner_coll", inner_task)
return Collection(
my_task, sub, auto_dash_names=auto_dash_names
)
def can_accept_collections_as_varargs_too(self):
sub = Collection("sub")
ns = Collection(sub)
assert ns.collections["sub"] == sub
def keys_dont_have_to_exist_in_full_path(self):
# Kinda duplicates earlier stuff; meh
# Key only stored on leaf
leaf = Collection("leaf", self.task)
leaf.configure({"key": "leaf-value"})
middle = Collection("middle", leaf)
root = Collection("root", middle)
config = root.configuration("middle.leaf.task")
assert config == {"key": "leaf-value"}
# Key stored on mid + leaf but not root
middle.configure({"key": "whoa"})
assert root.configuration("middle.leaf.task") == {"key": "whoa"}
def raises_ValueError_if_collection_named_same_as_task(self):
self.c.add_task(_mytask, "sub")
with raises(ValueError):
self.c.add_collection(Collection("sub"))
def invalid_subcollection_paths_result_in_KeyError(self):
# Straight up invalid
with raises(KeyError):
Collection("meh").configuration("nope.task")
# Exists but wrong level (should be 'root.task', not just
# 'task')
inner = Collection("inner", self.task)
with raises(KeyError):
Collection("root", inner).configuration("task")
def instantiate(obj_name=None):
# Explicitly given name wins over root ns name (if applicable),
# which wins over actual module name.
args = [name or obj_name or module_name]
kwargs = dict(
loaded_from=loaded_from, auto_dash_names=auto_dash_names
)
instance = cls(*args, **kwargs)
instance.__doc__ = module.__doc__
return instance
# See if the module provides a default NS to use in lieu of creating
# our own collection.
for candidate in ("ns", "namespace"):
obj = getattr(module, candidate, None)
if obj and isinstance(obj, Collection):
# TODO: make this into Collection.clone() or similar?
ret = instantiate(obj_name=obj.name)
ret.tasks = ret._transform_lexicon(obj.tasks)
ret.collections = ret._transform_lexicon(obj.collections)
ret.default = ret.transform(obj.default)
# Explicitly given config wins over root ns config
obj_config = copy_dict(obj._configuration)
if config:
merge_dicts(obj_config, config)
ret._configuration = obj_config
return ret
# Failing that, make our own collection from the module's tasks.
tasks = filter(lambda x: isinstance(x, Task), vars(module).values())
# Again, explicit name wins over implicit one from module path
collection = instantiate()
for task in tasks:
def add_collection(self, coll, name=None):
"""
Add `.Collection` ``coll`` as a sub-collection of this one.
:param coll: The `.Collection` to add.
:param str name:
The name to attach the collection as. Defaults to the collection's
own internal name.
.. versionadded:: 1.0
"""
# Handle module-as-collection
if isinstance(coll, types.ModuleType):
coll = Collection.from_module(coll)
# Ensure we have a name, or die trying
name = name or coll.name
if not name:
raise ValueError("Non-root collections must have a name!")
name = self.transform(name)
# Test for conflict
if name in self.tasks:
err = (
"Name conflict: this collection has a task named {!r} already"
) # noqa
raise ValueError(err.format(name))
# Insert
self.collections[name] = coll