Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
self.logger = unittest.mock.Mock(spec=logging.Logger)
self.profile = Qt.QWebEngineProfile()
self.account_jid = aioxmpp.JID.fromstr("juliet@capulet.lit")
self.conversation_jid = aioxmpp.JID.fromstr("romeo@montague.lit")
self.page = conversation.MessageViewPage(
self.profile,
logging.getLogger(
".".join([__name__, type(self).__qualname__])
),
self.account_jid,
self.conversation_jid,
)
run_coroutine(self.page.ready_event.wait())
from aioxmpp.testutils import (
make_connected_client,
CoroutineMock,
run_coroutine,
make_listener,
)
from aioxmpp.e2etest import (
blocking_timed,
TestCase,
)
LOCAL_JID = aioxmpp.JID.fromstr("juliet@capulet.example/balcony")
PEER_JID = aioxmpp.JID.fromstr("romeo@montague.example")
class TestConversation(unittest.TestCase):
def setUp(self):
self.listener = unittest.mock.Mock()
self.cc = make_connected_client()
self.cc.send = CoroutineMock()
self.cc.local_jid = LOCAL_JID
self.svc = unittest.mock.Mock(["client", "_conversation_left",
"logger"])
self.svc.client = self.cc
self.c = p2p.Conversation(self.svc, PEER_JID)
for ev in ["on_message"]:
########################################################################
import unittest
import unittest.mock
import aioxmpp.disco
import aioxmpp.mdr.service as mdr_service
import aioxmpp.mdr.xso as mdr_xso
import aioxmpp.service
import aioxmpp.tracking
from aioxmpp.testutils import (
make_connected_client,
)
TEST_TO = aioxmpp.JID.fromstr("romeo@montague.example")
class TestDeliveryReceiptsService(unittest.TestCase):
def setUp(self):
self.cc = make_connected_client()
self.disco_svr = aioxmpp.DiscoServer(self.cc)
self.t = aioxmpp.tracking.MessageTracker()
self.s = mdr_service.DeliveryReceiptsService(self.cc, dependencies={
aioxmpp.DiscoServer: self.disco_svr,
})
self.msg = unittest.mock.Mock(spec=aioxmpp.Message)
self.msg.xep0184_request_receipt = False
self.msg.to = TEST_TO
self.msg.id_ = "foo"
def tearDown(self):
def configure(self, section):
super().configure(section)
self.__host = section.get("host")
self._domain = aioxmpp.JID.fromstr(section.get(
"domain",
self.__host
))
self.__port = section.getint("port")
self.__security_layer = aioxmpp.make_security_layer(
None,
anonymous="",
**configure_tls_config(
section
)
)
self._quirks = configure_quirks(section)
def _create_client(self, sender_id, api_key, loop=None) -> aioxmpp.Client:
xmpp_client = aioxmpp.Client(
local_jid=aioxmpp.JID.fromstr('%s@gcm.googleapis.com' % sender_id),
security_layer=aioxmpp.make_security_layer(api_key),
override_peer=[
(self.FCM_HOST, self.FCM_PORT,
aioxmpp.connector.XMPPOverTLSConnector())
],
loop=loop
)
xmpp_client.on_stream_established.connect(
lambda: self._wait_connection.set_result(True)
)
xmpp_client.on_stream_destroyed.connect(
self._on_stream_destroyed
)
xmpp_client.on_failure.connect(
lambda exc: self._wait_connection.set_exception(exc)
)
async def _process_item(self, state, item, fut):
domain, last_seen, is_muc = item
address = aioxmpp.JID(localpart=None, domain=domain, resource=None)
if not is_muc:
threshold = datetime.utcnow() - self.non_muc_rescan_delay
if last_seen is not None and last_seen >= threshold:
self.logger.debug(
"%s is not a MUC service and was scanned since %s; skipping"
" it in in this round",
domain, threshold
)
return
else:
self.logger.debug(
"%s is not a MUC service, but it was not successfully "
"scanned since %s; including it in this round",
domain, threshold,
)
def configure(self):
super().configure()
self.muc_jid = self.args.muc
if self.muc_jid is None:
try:
self.muc_jid = aioxmpp.JID.fromstr(
self.config.get("muc_config", "muc_jid")
)
except (configparser.NoSectionError,
configparser.NoOptionError):
self.muc_jid = aioxmpp.JID.fromstr(
input("MUC JID> ")
)
if url.path() != "/" or url.host() != "":
logger.debug("invalid host or path")
request.fail(Qt.QWebEngineUrlRequestJob.UrlNotFound)
return
args = urllib.parse.parse_qs(url.query())
if "account" not in args or "peer" not in args:
request.fail(Qt.QWebEngineUrlRequestJob.UrlInvalid)
return
try:
account_s, = args["account"]
account_address = aioxmpp.JID.fromstr(account_s)
peer_s, = args["peer"]
peer = aioxmpp.JID.fromstr(peer_s)
except ValueError:
request.fail(Qt.QWebEngineUrlRequestJob.UrlInvalid)
return
try:
nickname, = args["nick"]
except (KeyError, ValueError):
nickname = None
try:
account = self._accounts.lookup_jid(account_address)
except KeyError:
logger.debug("could not find account: %r", account_address)
request.fail(Qt.QWebEngineUrlRequestJob.UrlNotFound)
return
async def join_muc(self, party_id: str) -> None:
muc_jid = aioxmpp.JID.fromstr(
'Party-{}@muc.prod.ol.epicgames.com'.format(party_id)
)
nick = '{0}:{0}:{1}'.format(
self._remove_illegal_characters(self.client.user.display_name),
self.client.user.id,
self.xmpp_client.local_jid.resource
)
room, fut = self.muc_service.join(
muc_jid,
nick
)
room.on_message.connect(self.muc_on_message)
room.on_join.connect(self.muc_on_member_join)
room.on_enter.connect(self.muc_on_enter)
Returns:
list: a list of filtered events
"""
if category and not to:
msg_slice = itertools.islice(
(x for x in self.store if x[2] == category), limit
)
elif to and not category:
to = JID.fromstr(to)
msg_slice = itertools.islice(
(x for x in self.store if _agent_in_msg(to, x[1])), limit
)
elif to and category:
to = JID.fromstr(to)
msg_slice = itertools.islice(
(x for x in self.store if _agent_in_msg(to, x[1]) and x[2] == category),
limit,
)
else:
msg_slice = self.all(limit=limit)
return msg_slice
return list(msg_slice)[::-1]