<commit_before>
"""
Infrastructure code for testing Gabble by pretending to be a Jabber server.
"""

import pprint
import sys
import traceback

import dbus
import dbus.glib

from twisted.words.xish import xpath
from twisted.words.protocols.jabber.client import IQ
from twisted.words.protocols.jabber import xmlstream
from twisted.internet import reactor

tp_name_prefix = 'org.freedesktop.Telepathy'
tp_path_prefix = '/org/freedesktop/Telepathy'

class Authenticator(xmlstream.Authenticator):
    "Trivial XML stream authenticator that accepts one username/digest pair."

    def __init__(self, expected_username, expected_digest):
        self.expected_username = expected_username
        self.expected_digest = expected_digest
        xmlstream.Authenticator.__init__(self)

    def connectionMade(self):
        self.xmlstream.initiating = False
        self.xmlstream.sid = '1'
        self.xmlstream.sendHeader()

    def streamStarted(self):
        self.xmlstream.addOnetimeObserver("/iq", self.initialIq)

    def initialIq(self, iq):
        result = IQ(self.xmlstream, "result")
        result["id"] = iq["id"]
        query = result.addElement('query')
        query["xmlns"] = "jabber:iq:auth"
        query.addElement('username', content='test')
        query.addElement('password')
        query.addElement('digest')
        query.addElement('resource')
        self.xmlstream.addOnetimeObserver('/iq/query/username', self.secondIq)
        self.xmlstream.send(result)

    def secondIq(self, iq):
        username = xpath.queryForNodes('/iq/query/username', iq)
        assert map(str, username) == [self.expected_username]

        digest = xpath.queryForNodes('/iq/query/digest', iq)
        assert map(str, digest) == [self.expected_digest]

        resource = xpath.queryForNodes('/iq/query/resource', iq)
        assert map(str, resource) == ['Resource']

        result = IQ(self.xmlstream, "result")
        result["id"] = iq["id"]
        self.xmlstream.send(result)
        self.xmlstream.dispatch(self.xmlstream, xmlstream.STREAM_AUTHD_EVENT)

class XmlStream(xmlstream.XmlStream):
    def __init__(self, handler, authenticator):
        xmlstream.XmlStream.__init__(self, authenticator)
        self.handler = handler
        handler.data['stream'] = self
        self.addObserver('//iq', lambda x: handler.handle_event(
            ('stream-iq', x)))
        self.addObserver('//message', lambda x: handler.handle_event(
            ('stream-message', x)))
        self.addObserver('//presence', lambda x: handler.handle_event(
            ('stream-presence', x)))
        self.addObserver('//event/stream/authd', self._cb_authd)

    def _cb_authd(self, _):
        # caleld when stream is authenticated
        self.addObserver(
            "/iq/query[@xmlns='http://jabber.org/protocol/disco#info']",
            self._cb_disco_iq)
        self.handler.handle_event(('stream-authenticated',))

    def _cb_disco_iq(self, iq):
        # no features
        iq['type'] = 'result'
        self.send(iq)

class XmlStreamFactory(xmlstream.XmlStreamFactory):
    def __init__(self, handler, authenticator):
        xmlstream.XmlStreamFactory.__init__(self, authenticator)
        self.handler = handler

    def buildProtocol(self, _):
        # XXX: This is necessary because xmlstream.XmlStreamFactory's
        # buildProtocol doesn't honour self.protocol. This is fixed in SVN.<commit_msg>gabbletest.py: only respond to disco request if host is localhost, so that we can emulate conference servers<commit_after>
"""
Infrastructure code for testing Gabble by pretending to be a Jabber server.
"""

import pprint
import sys
import traceback

import dbus
import dbus.glib

from twisted.words.xish import xpath
from twisted.words.protocols.jabber.client import IQ
from twisted.words.protocols.jabber import xmlstream
from twisted.internet import reactor

tp_name_prefix = 'org.freedesktop.Telepathy'
tp_path_prefix = '/org/freedesktop/Telepathy'

class Authenticator(xmlstream.Authenticator):
    "Trivial XML stream authenticator that accepts one username/digest pair."

    def __init__(self, expected_username, expected_digest):
        self.expected_username = expected_username
        self.expected_digest = expected_digest
        xmlstream.Authenticator.__init__(self)

    def connectionMade(self):
        self.xmlstream.initiating = False
        self.xmlstream.sid = '1'
        self.xmlstream.sendHeader()

    def streamStarted(self):
        self.xmlstream.addOnetimeObserver("/iq", self.initialIq)

    def initialIq(self, iq):
        result = IQ(self.xmlstream, "result")
        result["id"] = iq["id"]
        query = result.addElement('query')
        query["xmlns"] = "jabber:iq:auth"
        query.addElement('username', content='test')
        query.addElement('password')
        query.addElement('digest')
        query.addElement('resource')
        self.xmlstream.addOnetimeObserver('/iq/query/username', self.secondIq)
        self.xmlstream.send(result)

    def secondIq(self, iq):
        username = xpath.queryForNodes('/iq/query/username', iq)
        assert map(str, username) == [self.expected_username]

        digest = xpath.queryForNodes('/iq/query/digest', iq)
        assert map(str, digest) == [self.expected_digest]

        resource = xpath.queryForNodes('/iq/query/resource', iq)
        assert map(str, resource) == ['Resource']

        result = IQ(self.xmlstream, "result")
        result["id"] = iq["id"]
        self.xmlstream.send(result)
        self.xmlstream.dispatch(self.xmlstream, xmlstream.STREAM_AUTHD_EVENT)

class XmlStream(xmlstream.XmlStream):
    def __init__(self, handler, authenticator):
        xmlstream.XmlStream.__init__(self, authenticator)
        self.handler = handler
        handler.data['stream'] = self
        self.addObserver('//iq', lambda x: handler.handle_event(
            ('stream-iq', x)))
        self.addObserver('//message', lambda x: handler.handle_event(
            ('stream-message', x)))
        self.addObserver('//presence', lambda x: handler.handle_event(
            ('stream-presence', x)))
        self.addObserver('//event/stream/authd', self._cb_authd)

    def _cb_authd(self, _):
        # caleld when stream is authenticated
        self.addObserver(
            "/iq/query[@xmlns='http://jabber.org/protocol/disco#info']",
            self._cb_disco_iq)
        self.handler.handle_event(('stream-authenticated',))

    def _cb_disco_iq(self, iq):
        if iq['to'] == 'localhost':
            # no features
            iq['type'] = 'result'
            self.send(iq)

class XmlStreamFactory(xmlstream.XmlStreamFactory):
    def __init__(self, handler, authenticator):
        xmlstream.XmlStreamFactory.__init__(self, authenticator)
        self.handler = handler

    def buildProtocol(self, _):
        # XXX: This is necessary because xmlstream.XmlStreamFactory's
        # buildProtocol doesn't honour self.protocol. This is fixed in SVN.
{'fraction_non_alphanumeric': 0.11864163920332425, 'fraction_numerical': 0.0002865740077374982, 'mean_word_length': 3.1350710900473935, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 44}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2929308', 'n_tokens_mistral': 2101, 'n_tokens_neox': 2020, 'n_words': 426}