Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_options(self):
self.assertIsInstance(
pubsub_xso.Request.options,
xso.Child
)
self.assertSetEqual(
pubsub_xso.Request.options._classes,
{
pubsub_xso.Options
}
def test_inheritance_child_ambiguous(self):
class ClsLeafA:
TAG = "foo"
class ClsA(metaclass=xso_model.XMLStreamClass):
c1 = xso.Child([ClsLeafA])
with self.assertRaisesRegexp(TypeError, "ambiguous Child properties"):
class ClsB(ClsA):
c2 = xso.Child([ClsLeafA])
def test_from_events(self):
dest = []
def mock_fun(ev_args, ctx):
dest.append(ev_args)
while True:
value = yield
dest.append(value)
if value[0] == "stop":
return "bar"
Cls = unittest.mock.MagicMock()
Cls.TAG = None, "foo"
Cls.parse_events = mock_fun
prop = xso.Child([Cls])
instance = make_instance_mock()
gen = prop.from_events(instance, (None, "foo", {}), self.ctx)
next(gen)
with self.assertRaises(StopIteration) as ctx:
gen.send(("stop",))
self.assertEqual(
"bar",
ctx.exception.value
)
self.assertSequenceEqual(
[
(None, "foo", {}),
("stop",)
],
lang = xso.Attr(
tag=(namespaces.xml, "lang"),
default=None
)
child = xso.Child([Baz])
class Foo(xso.XSO):
TAG = "foo"
lang = xso.Attr(
tag=(namespaces.xml, "lang"),
missing=xso.lang_attr
)
child = xso.Child([Bar])
ctx = xso_model.Context()
caught_obj = None
def catch(obj):
nonlocal caught_obj
caught_obj = obj
sd = xso.SAXDriver(
functools.partial(from_wrapper,
Foo.parse_events,
ctx=ctx),
on_emit=catch
)
lxml.sax.saxify(
etree.fromstring(
TAG = "bar"
lang = xso.Attr(
tag=(namespaces.xml, "lang"),
missing=xso.lang_attr
)
class Foo(xso.XSO):
TAG = "foo"
lang = xso.Attr(
tag=(namespaces.xml, "lang"),
missing=xso.lang_attr
)
child = xso.Child([Bar])
ctx = xso_model.Context()
ctx.lang = "de-DE"
caught_obj = None
def catch(obj):
nonlocal caught_obj
caught_obj = obj
sd = xso.SAXDriver(
functools.partial(from_wrapper,
Foo.parse_events,
ctx=ctx),
on_emit=catch
)
lxml.sax.saxify(
def test_forbid_ambiguous_children(self):
class ClsA(metaclass=xso_model.XMLStreamClass):
TAG = "foo"
class ClsB(metaclass=xso_model.XMLStreamClass):
TAG = "foo"
with self.assertRaisesRegexp(TypeError, "ambiguous Child properties"):
class Cls(metaclass=xso_model.XMLStreamClass):
c1 = xso.Child([ClsA])
c2 = xso.Child([ClsB])
def test_actor(self):
self.assertIsInstance(
muc_xso.AdminItem.actor,
xso.Child
)
self.assertSetEqual(
muc_xso.AdminItem.actor._classes,
{muc_xso.AdminActor}
)
def test_actions_attr(self):
self.assertIsInstance(
adhoc_xso.Command.actions,
xso.Child,
)
self.assertCountEqual(
adhoc_xso.Command.actions._classes,
[
adhoc_xso.Actions,
]
id_ = xso.Attr(tag="id")
type_ = xso.Attr(
tag="type",
type_=xso.EnumCDataType(
structs.IQType,
allow_coerce=True,
deprecate_coerce=True,
# changing the following breaks stanza handling; StanzaStream
# relies on the meta-properties of the enumerations (is_request and
# such).
allow_unknown=False,
accept_unknown=False,
)
)
payload = xso.Child([], strict=True)
def __init__(self, type_, *, payload=None, error=None, **kwargs):
super().__init__(**kwargs)
self.type_ = type_
self.payload = payload
self.error = error
def _validate(self):
try:
self.id_
except AttributeError:
raise ValueError("IQ requires ID") from None
if self.type_ == structs.IQType.ERROR and self.error is None:
raise ValueError("IQ with type='error' requires error payload")
DECLARE_NS = {}
from_ = xso.Attr(
tag="from",
type_=xso.JID(),
default=None)
to = xso.Attr(
tag="to",
type_=xso.JID(),
default=None)
lang = xso.LangAttr(
tag=(namespaces.xml, "lang")
)
error = xso.Child([Error])
def __init__(self, *, from_=None, to=None, id_=None):
super().__init__()
if from_ is not None:
self.from_ = from_
if to is not None:
self.to = to
if id_ is not None:
self.id_ = id_
def autoset_id(self):
"""
If the :attr:`id_` already has a non-false (false is also the empty
string!) value, this method is a no-op.
Otherwise, the :attr:`id_` attribute is filled with eight bytes of