Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
pass
class ResourceB(Resource):
pass
class DriverWithA(Driver):
bindings = {"res": ResourceA}
class DriverWithASet(Driver):
bindings = {"res": {ResourceA}, }
class DriverWithAB(Driver):
bindings = {"res": {ResourceA, ResourceB}, }
def test_suppliers_a(target):
ra = ResourceA(target, "resource")
d = DriverWithA(target, "resource")
assert d.res is ra
def test_suppliers_aset(target):
ra = ResourceA(target, "resource")
d = DriverWithASet(target, "driver")
assert d.res is ra
def test_suppliers_ab_a(target):
class DriverWithNamedA(Driver):
bindings = {
"res": Driver.NamedBinding(ResourceA),
}
def test_suppliers_named_a(target):
ra = ResourceA(target, "resource")
target.set_binding_map({"res": "resource"})
d = DriverWithNamedA(target, "driver")
assert d.res is ra
class DriverWithMultiA(Driver):
bindings = {
"res1": ResourceA,
"res2": ResourceA,
}
def test_suppliers_multi_a(target):
ra1 = ResourceA(target, "resource1")
with pytest.raises(BindingError) as excinfo:
DriverWithMultiA(target, "driver")
assert "duplicate bindings" in excinfo.value.msg
def test_suppliers_multi_a_explict(target):
ra1 = ResourceA(target, "resource1")
ra2 = ResourceA(target, "resource2")
def test_getitem(target):
class AProtocol(abc.ABC):
pass
class A(Driver, AProtocol):
pass
class B(Driver):
pass
a = A(target, "adriver")
target.activate(a)
assert isinstance(target[A], A)
assert target[A] is a
assert target[AProtocol] is a
assert target[A, "adriver"] is a
assert target[AProtocol, "adriver"] is a
with pytest.raises(NoDriverFoundError) as excinfo:
target[A, "bdriver"]
assert "matching resources with other names" in excinfo.value.msg
with pytest.raises(NoDriverFoundError) as excinfo:
target[B, "adriver"]
assert re.match("no active .*? driver named '{}' found in Target".format(a.name),
excinfo.value.msg)
assert "duplicate bindings" in excinfo.value.msg
def test_suppliers_multi_a_explict(target):
ra1 = ResourceA(target, "resource1")
ra2 = ResourceA(target, "resource2")
target.set_binding_map({
"res1": "resource1",
"res2": "resource2",
})
d = DriverWithMultiA(target, "driver")
assert d.res1 is ra1
assert d.res2 is ra2
class DriverWithNamedMultiA(Driver):
bindings = {
"res1": Driver.NamedBinding(ResourceA),
"res2": Driver.NamedBinding(ResourceA),
}
def test_suppliers_multi_named_a(target):
ra1 = ResourceA(target, "resource1")
ra2 = ResourceA(target, "resource2")
target.set_binding_map({
"res1": "resource1",
"res2": "resource2",
})
d = DriverWithNamedMultiA(target, "driver")
assert d.res1 is ra1
assert d.res2 is ra2
def test_suppliers_multi_named_a(target):
ra1 = ResourceA(target, "resource1")
ra2 = ResourceA(target, "resource2")
target.set_binding_map({
"res1": "resource1",
"res2": "resource2",
})
d = DriverWithNamedMultiA(target, "driver")
assert d.res1 is ra1
assert d.res2 is ra2
# test optional bindings
class DriverWithOptionalA(Driver):
bindings = {"res": {ResourceA, None}, }
class DriverWithOptionalAB(Driver):
bindings = {"res": {ResourceA, ResourceB, None}, }
def test_suppliers_optional_a(target):
ra = ResourceA(target, "resource")
d = DriverWithOptionalA(target, "driver")
assert d.res is ra
def test_suppliers_optional_a_missing(target):
rb = ResourceB(target, "resource")
d = DriverWithOptionalA(target, "driver")
def test_suppliers_ab_missing(target):
with pytest.raises(NoSupplierFoundError):
d = DriverWithAB(target, "driver")
def test_suppliers_unexpected_binding(target):
ra = ResourceA(target, "resource")
target.set_binding_map({"res": "resource", "unexpected": "foo"})
with pytest.raises(BindingError) as excinfo:
DriverWithA(target, "driver")
assert "got unexpected bindings" in excinfo.value.msg
class DriverWithNamedA(Driver):
bindings = {
"res": Driver.NamedBinding(ResourceA),
}
def test_suppliers_named_a(target):
ra = ResourceA(target, "resource")
target.set_binding_map({"res": "resource"})
d = DriverWithNamedA(target, "driver")
assert d.res is ra
class DriverWithMultiA(Driver):
bindings = {
"res1": ResourceA,
"res2": ResourceA,
def test_suppliers_optional_a_missing(target):
rb = ResourceB(target, "resource")
d = DriverWithOptionalA(target, "driver")
assert d.res is None
def test_suppliers_optional_ab_a(target):
ra = ResourceA(target, "resource")
d = DriverWithOptionalAB(target, "driver")
assert d.res is ra
class DriverWithOptionalNamedA(Driver):
bindings = {
"res": Driver.NamedBinding({ResourceA, None}),
}
def test_suppliers_optional_named_a(target):
ra = ResourceA(target, "resource")
target.set_binding_map({"res": "resource"})
d = DriverWithOptionalNamedA(target, "driver")
assert d.res is ra
def test_suppliers_optional_named_a_missing(target):
rb = ResourceB(target, "resource")
target.set_binding_map({"res": "resource"})
d = DriverWithOptionalNamedA(target, "driver")
assert d.res is None
import attr
from ..binding import BindingError
from ..driver import Driver
@attr.s(eq=False)
class StrategyError(Exception):
msg = attr.ib(validator=attr.validators.instance_of(str))
@attr.s(eq=False)
class Strategy(Driver): # reuse driver handling
"""
Represents a strategy which places a target into a requested state by
calling specific drivers. A strategy usually needs to know some details of
a given target.
Life cycle:
- create
- bind (n times)
- usage
TODO: This might also be just a driver?
"""
def __attrs_post_init__(self):
super().__attrs_post_init__()
if self.target is None:
>>> target = Target('main')
>>> console = FakeConsoleDriver(target, 'console')
>>> target.activate(console)
>>> target[FakeConsoleDriver]
FakeConsoleDriver(target=Target(name='main', …), name='console', …)
>>> target[FakeConsoleDriver, 'console']
FakeConsoleDriver(target=Target(name='main', …), name='console', …)
"""
name = None
if not isinstance(key, tuple):
cls = key
elif len(key) == 2:
cls, name = key
if isinstance(cls, str):
cls = self._reg_class_from_string(cls)
if not issubclass(cls, (Driver, abc.ABC)): # all Protocols derive from ABC
raise NoDriverFoundError(
"invalid driver class {}".format(cls)
)
return self.get_active_driver(cls, name=name)
"{} is not in state {}".format(client, BindingState.idle)
)
# consistency check
assert isinstance(client, Driver)
assert client not in self.drivers
assert client.target is None
mapping = self._binding_map
self._binding_map = {}
# locate suppliers
bound_suppliers = []
for name, requirements in client.bindings.items():
explicit = False
if isinstance(requirements, Driver.NamedBinding):
requirements = requirements.value
explicit = True
supplier_name = mapping.pop(name, None)
if explicit and supplier_name is None:
raise BindingError(
"supplier for {name} ({requirements}) of {driver} in {target} requires an explicit name".format( # pylint: disable=line-too-long
name=name, requirements=requirements, driver=client, target=self)
)
# use sets even for a single requirement and make a local copy
if not isinstance(requirements, set):
requirements = {requirements}
else:
requirements = requirements.copy()
# None indicates that the binding is optional
optional = None in requirements
requirements.discard(None)