forked from rc/aircox
write sound monitor tests
This commit is contained in:
100
aircox/test.py
100
aircox/test.py
@ -44,38 +44,64 @@ InterfaceTarget = namedtuple(
|
||||
|
||||
|
||||
class WrapperMixin:
|
||||
def __init__(self, target=None, ns=None, ns_attr=None, **kwargs):
|
||||
type_interface = None
|
||||
"""For instance of class wrapped by an Interface, this is the wrapping
|
||||
interface of the class."""
|
||||
instances = None
|
||||
ns = None
|
||||
ns_attr = None
|
||||
|
||||
def __init__(
|
||||
self, target=None, ns=None, ns_attr=None, type_interface=None, **kwargs
|
||||
):
|
||||
self.target = target
|
||||
if ns:
|
||||
self.inject(ns, ns_attr)
|
||||
if self.type_interface:
|
||||
self._set_type_interface(type_interface)
|
||||
super().__init__(**kwargs)
|
||||
|
||||
def _set_type_interface(self, type_interface):
|
||||
if self.type_interface:
|
||||
raise RuntimeError("a type interface is already assigned")
|
||||
|
||||
self.type_interface = type_interface
|
||||
if not type_interface.instances:
|
||||
type_interface.instances = [self]
|
||||
else:
|
||||
type_interface.instances.append(self)
|
||||
|
||||
@property
|
||||
def ns_target(self):
|
||||
"""Actual namespace's target (using ns.ns_attr)"""
|
||||
if self.ns and self.ns_attr:
|
||||
return getattr(self.ns, self.ns_attr, None)
|
||||
return None
|
||||
|
||||
def inject(self, ns=None, ns_attr=None):
|
||||
if ns and ns_attr:
|
||||
ns_target = getattr(ns, ns_attr, None)
|
||||
if self.target is ns_target:
|
||||
return
|
||||
elif self.target is not None:
|
||||
raise RuntimeError(
|
||||
"self target already injected. It must be "
|
||||
"`release` before `inject`."
|
||||
)
|
||||
self.target = ns_target
|
||||
setattr(ns, ns_attr, self.parent)
|
||||
elif not ns or not ns_attr:
|
||||
"""Inject interface into namespace at given key."""
|
||||
if not (ns and ns_attr):
|
||||
raise ValueError("ns and ns_attr must be provided together")
|
||||
|
||||
ns_target = getattr(ns, ns_attr, None)
|
||||
if self.target is ns_target:
|
||||
return
|
||||
elif self.target is not None:
|
||||
raise RuntimeError(
|
||||
"self target already injected. It must be "
|
||||
"`release` before `inject`."
|
||||
)
|
||||
|
||||
self.target = ns_target
|
||||
setattr(ns, ns_attr, self.interface)
|
||||
|
||||
self.ns = ns
|
||||
self.ns_attr = ns_attr
|
||||
|
||||
def release(self):
|
||||
if self.ns_target is self:
|
||||
setattr(self.target.namespace, self.target.name, self.target)
|
||||
"""Remove injection from previously injected parent, reset target."""
|
||||
if self.ns_target is self.interface:
|
||||
setattr(self.ns, self.ns_attr, self.target)
|
||||
self.target = None
|
||||
|
||||
|
||||
@ -83,7 +109,9 @@ class SpoofMixin:
|
||||
traces = None
|
||||
|
||||
def __init__(self, funcs=None, **kwargs):
|
||||
self.reset(funcs or {})
|
||||
self.reset(
|
||||
funcs or {},
|
||||
)
|
||||
super().__init__(**kwargs)
|
||||
|
||||
def reset(self, funcs=None):
|
||||
@ -152,23 +180,19 @@ class SpoofMixin:
|
||||
"""
|
||||
func = self.funcs[name]
|
||||
if callable(func):
|
||||
return func(*a, **kw)
|
||||
return func(self, *a, **kw)
|
||||
return func
|
||||
|
||||
|
||||
class InterfaceMeta(SpoofMixin, WrapperMixin):
|
||||
calls = None
|
||||
"""Calls done."""
|
||||
|
||||
def __init__(self, parent, **kwargs):
|
||||
self.parent = parent
|
||||
super().__init__(**kwargs)
|
||||
|
||||
def __getitem__(self, name):
|
||||
return self.traces[name]
|
||||
|
||||
|
||||
class Interface:
|
||||
class IMeta(SpoofMixin, WrapperMixin):
|
||||
def __init__(self, interface, **kwargs):
|
||||
self.interface = interface
|
||||
super().__init__(**kwargs)
|
||||
|
||||
def __getitem__(self, name):
|
||||
return self.traces[name]
|
||||
|
||||
_imeta = None
|
||||
"""This contains a InterfaceMeta instance related to Interface one.
|
||||
|
||||
@ -182,7 +206,7 @@ class Interface:
|
||||
_imeta_kw.setdefault("funcs", _funcs)
|
||||
if _target is not None:
|
||||
_imeta_kw.setdefault("target", _target)
|
||||
self._imeta = InterfaceMeta(self, **_imeta_kw)
|
||||
self._imeta = self.IMeta(self, **_imeta_kw)
|
||||
self.__dict__.update(kwargs)
|
||||
|
||||
@property
|
||||
@ -195,19 +219,29 @@ class Interface:
|
||||
return cls(**kwargs)
|
||||
|
||||
def _irelease(self):
|
||||
"""Shortcut to `self._imeta.release`."""
|
||||
self._imeta.release()
|
||||
|
||||
def _trace(self, *args, **kw):
|
||||
"""Shortcut to `self._imeta.get_trace`."""
|
||||
return self._imeta.get_trace(*args, **kw)
|
||||
|
||||
def _traces(self, *args, **kw):
|
||||
"""Shortcut to `self._imeta.get_traces`."""
|
||||
return self._imeta.get_traces(*args, **kw)
|
||||
|
||||
def __call__(self, *args, **kwargs):
|
||||
target = self._imeta.target
|
||||
print("is it class?", self, target, inspect.isclass(target))
|
||||
if inspect.isclass(target):
|
||||
target = target(*args, **kwargs)
|
||||
return type(self)(target, _imeta_kw={"funcs": self._imeta.funcs})
|
||||
return type(self)(
|
||||
target,
|
||||
_imeta_kw={"type_interface": self, "funcs": self._imeta.funcs},
|
||||
)
|
||||
|
||||
if "__call__" in self._imeta.funcs:
|
||||
return self._imeta.call("__call__", args, kwargs)
|
||||
|
||||
self._imeta.add("__call__", args, kwargs)
|
||||
return self._imeta.target(*args, **kwargs)
|
||||
@ -216,3 +250,7 @@ class Interface:
|
||||
if attr in self._imeta.funcs:
|
||||
return lambda *args, **kwargs: self._imeta.call(attr, args, kwargs)
|
||||
return getattr(self._imeta.target, attr)
|
||||
|
||||
def __str__(self):
|
||||
iface = super().__str__()
|
||||
return f"{iface}::{self._imeta.target}"
|
||||
|
Reference in New Issue
Block a user