Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import unittest
import logging
from slixmpp.plugins.base import PluginManager, BasePlugin, register_plugin
class A(BasePlugin):
name = 'a'
class B(BasePlugin):
name = 'b'
class C(BasePlugin):
name = 'c'
dependencies = {'b', 'd'}
class D(BasePlugin):
name = 'd'
dependencies = {'c'}
class E(BasePlugin):
name = 'e'
dependencies = {'a', 'd'}
class A(BasePlugin):
name = 'a'
class B(BasePlugin):
name = 'b'
class C(BasePlugin):
name = 'c'
dependencies = {'b', 'd'}
class D(BasePlugin):
name = 'd'
dependencies = {'c'}
class E(BasePlugin):
name = 'e'
dependencies = {'a', 'd'}
class F(BasePlugin):
name = 'f'
dependencies = {'a', 'b'}
register_plugin(A)
register_plugin(B)
register_plugin(C)
import unittest
import logging
from slixmpp.plugins.base import PluginManager, BasePlugin, register_plugin
class A(BasePlugin):
name = 'a'
class B(BasePlugin):
name = 'b'
class C(BasePlugin):
name = 'c'
dependencies = {'b', 'd'}
class D(BasePlugin):
name = 'd'
dependencies = {'c'}
See the file LICENSE for copying permission.
"""
import logging
from slixmpp.stanza import StreamFeatures
from slixmpp.features.feature_preapproval import stanza
from slixmpp.xmlstream import register_stanza_plugin
from slixmpp.plugins.base import BasePlugin
log = logging.getLogger(__name__)
class FeaturePreApproval(BasePlugin):
name = 'feature_preapproval'
description = 'RFC 6121: Stream Feature: Subscription Pre-Approval'
dependences = set()
stanza = stanza
def plugin_init(self):
self.xmpp.register_feature('preapproval',
self._handle_preapproval,
restart=False,
order=9001)
register_stanza_plugin(StreamFeatures, stanza.PreApproval)
def _handle_preapproval(self, features):
"""Save notice that the server support subscription pre-approvals.
def post_init(self):
""" Init complete. Register our features in Service discovery. """
BasePlugin.post_init(self)
self.xmpp['xep_0030'].add_feature(Sensordata.namespace)
self.xmpp['xep_0030'].set_items(node=Sensordata.namespace, items=tuple())
See the file LICENSE for copying permission.
"""
import logging
from slixmpp.stanza import StreamFeatures
from slixmpp.features.feature_rosterver import stanza
from slixmpp.xmlstream import register_stanza_plugin
from slixmpp.plugins.base import BasePlugin
log = logging.getLogger(__name__)
class FeatureRosterVer(BasePlugin):
name = 'feature_rosterver'
description = 'RFC 6121: Stream Feature: Roster Versioning'
dependences = set()
stanza = stanza
def plugin_init(self):
self.xmpp.register_feature('rosterver',
self._handle_rosterver,
restart=False,
order=9000)
register_stanza_plugin(StreamFeatures, stanza.RosterVer)
def _handle_rosterver(self, features):
"""Enable using roster versioning.
import datetime
from threading import Thread, Lock, Timer
from slixmpp.plugins.xep_0323.timerreset import TimerReset
from slixmpp.xmlstream import JID
from slixmpp.xmlstream.handler import Callback
from slixmpp.xmlstream.matcher import StanzaPath
from slixmpp.plugins.base import BasePlugin
from slixmpp.plugins.xep_0323 import stanza
from slixmpp.plugins.xep_0323.stanza import Sensordata
log = logging.getLogger(__name__)
class XEP_0323(BasePlugin):
"""
XEP-0323: IoT Sensor Data
This XEP provides the underlying architecture, basic operations and data
structures for sensor data communication over XMPP networks. It includes
a hardware abstraction model, removing any technical detail implemented
in underlying technologies.
Also see
Configuration Values:
threaded -- Indicates if communication with sensors should be threaded.
Defaults to True.
Slixmpp: The Slick XMPP Library
Copyright (C) 2013 Nathanael C. Fritz, Lance J.T. Stout
This file is part of Slixmpp.
See the file LICENSE for copying permission.
"""
from slixmpp.plugins.base import register_plugin, BasePlugin
from slixmpp.plugins.google.gmail import Gmail
from slixmpp.plugins.google.auth import GoogleAuth
from slixmpp.plugins.google.settings import GoogleSettings
from slixmpp.plugins.google.nosave import GoogleNoSave
class Google(BasePlugin):
"""
Google: Custom GTalk Features
Also see:
"""
name = 'google'
description = 'Google: Custom GTalk Features'
dependencies = set([
'gmail',
'google_settings',
'google_nosave',
'google_auth'
])
Copyright (C) 2011 Nathanael C. Fritz, Lance J.T. Stout
This file is part of Slixmpp.
See the file LICENSE for copying permission.
"""
import logging
from slixmpp.plugins.base import BasePlugin
from slixmpp.plugins.xep_0196 import stanza, UserGaming
log = logging.getLogger(__name__)
class XEP_0196(BasePlugin):
"""
XEP-0196: User Gaming
"""
name = 'xep_0196'
description = 'XEP-0196: User Gaming'
dependencies = {'xep_0163'}
stanza = stanza
def plugin_end(self):
self.xmpp['xep_0030'].del_feature(feature=UserGaming.namespace)
self.xmpp['xep_0163'].remove_interest(UserGaming.namespace)
def session_bind(self, jid):
self.xmpp['xep_0163'].register_pep('user_gaming', UserGaming)
import time
from slixmpp import asyncio
from functools import partial
from slixmpp.xmlstream import JID
from slixmpp.xmlstream.handler import Callback
from slixmpp.xmlstream.matcher import StanzaPath
from slixmpp.plugins.base import BasePlugin
from slixmpp.plugins.xep_0325 import stanza
from slixmpp.plugins.xep_0325.stanza import Control
log = logging.getLogger(__name__)
class XEP_0325(BasePlugin):
"""
XEP-0325: IoT Control
Actuators are devices in sensor networks that can be controlled through
the network and act with the outside world. In sensor networks and
Internet of Things applications, actuators make it possible to automate
real-world processes.
This plugin implements a mechanism whereby actuators can be controlled
in XMPP-based sensor networks, making it possible to integrate sensors
and actuators of different brands, makes and models into larger
Internet of Things applications.
Also see