<commit_before>"""
Receives several tube offers:
- Test to accept a 1-1 stream tube
  - using UNIX sockets and IPv4 sockets
  - using the old tube iface and the new tube iface
- Test to accept with bad parameters
- Test to refuse the tube offer
  - using the old tube iface and the new tube iface
"""

import dbus

from servicetest import call_async, EventPattern, EventProtocolClientFactory
from gabbletest import acknowledge_iq

from twisted.words.xish import domish, xpath
from twisted.internet import reactor
import ns
import constants as cs
from bytestream import create_from_si_offer
import tubetestutil as t

bob_jid = 'bob@localhost/Bob'
stream_tube_id = 49

def receive_tube_offer(q, bus, conn, stream):
    message = domish.Element(('jabber:client', 'message'))
    message['to'] = 'test@localhost/Resource'
    message['from'] = bob_jid
    tube_node = message.addElement((ns.TUBES, 'tube'))
    tube_node['type'] = 'stream'
    tube_node['service'] = 'http'
    tube_node['id'] = str(stream_tube_id)
    stream.send(message)

    old_sig, new_sig = q.expect_many(
        EventPattern('dbus-signal', signal='NewChannel'),
        EventPattern('dbus-signal', signal='NewChannels'),
        )
    chan_path = old_sig.args[0]
    assert old_sig.args[1] == cs.CHANNEL_TYPE_TUBES, old_sig.args[1]
    assert old_sig.args[2] == cs.HT_CONTACT
    bob_handle = old_sig.args[3]
    assert old_sig.args[2] == 1, old_sig.args[2] # Suppress_Handler
    assert len(new_sig.args) == 1
    assert len(new_sig.args[0]) == 1
    assert new_sig.args[0][0][0] == chan_path, new_sig.args[0][0]
    assert new_sig.args[0][0][1] is not None

    event = q.expect('dbus-signal', signal='NewTube')

    old_sig, new_sig = q.expect_many(
        EventPattern('dbus-signal', signal='NewChannel'),
        EventPattern('dbus-signal', signal='NewChannels'),
        )
    new_chan_path = old_sig.args[0]
    assert new_chan_path != chan_path
    assert old_sig.args[1] == cs.CHANNEL_TYPE_STREAM_TUBE, old_sig.args[1]
    assert old_sig.args[2] == cs.HT_CONTACT
    bob_handle = old_sig.args[3]
    assert old_sig.args[2] == 1, old_sig.args[2] # Suppress_Handler
    assert len(new_sig.args) == 1
    assert len(new_sig.args[0]) == 1
    assert new_sig.args[0][0][0] == new_chan_path, new_sig.args[0][0]
    assert new_sig.args[0][0][1] is not None

    # create channel proxies
    tubes_chan = bus.get_object(conn.bus_name, chan_path)
    tubes_iface = dbus.Interface(tubes_chan, cs.CHANNEL_TYPE_TUBES)

    new_tube_chan = bus.get_object(conn.bus_name, new_chan_path)
    new_tube_iface = dbus.Interface(new_tube_chan, cs.CHANNEL_TYPE_STREAM_TUBE)

    return (tubes_chan, tubes_iface, new_tube_chan, new_tube_iface)

def expect_tube_activity(q, bus, conn, stream, bytestream_cls):
    event_socket, event_iq = q.expect_many(
            EventPattern('socket-connected'),
            EventPattern('stream-iq', to=bob_jid, query_ns=ns.SI,
                query_name='si'))

    protocol = event_socket.protocol
    protocol.sendData("hello initiator")

    bytestream, profile = create_from_si_offer(stream, q, bytestream_cls, event_iq.stanza,
        'test@localhost/Resource')

    assert profile == ns.TUBES

    stream_node = xpath.queryForNodes('/iq/si/stream[@xmlns="%s"]' %
        ns.TUBES, event_iq.stanza)[0]
    assert stream_node is not None
    assert stream_node['tube'] == str(stream_tube_id)

    result, si = bytestream.create_si_reply(event_iq.stanza)
    si.addElement((ns.TUBES, 'tube'))
    stream.send(result)

    bytestream.wait_bytestream_open()

    data = bytestream.get_data()
    assert data == 'hello initiator'

    # reply to the initiator
    bytestream.send_data('hello joiner')

    e = q.expect('socket-data')
    assert e.data == 'hello joiner'

    return bytestream

def test(q, bus, conn, stream, bytestream_cls):
    conn.Connect()

    _, vcard_event, roster_event = q.expect_many(
        EventPattern('dbus-signal', signal='StatusChanged', args=[0, 1]),
        EventPattern('stream-iq', to=None, query_ns='vcard-temp',
            query_name='vCard'),
        EventPattern('stream-iq', query_ns='jabber:iq:roster'))

    acknowledge_iq(stream, vcard_event.stanza)

    roster = roster_event.stanza
    roster['type'] = 'result'
    item = roster_event.query.addElement('item')
    item['jid'] = 'bob@localhost' # Bob can do tubes
    item['subscription'] = 'both'
    stream.send(roster)

    # Send Bob presence and his caps
    presence = domish.Element(('jabber:client', 'presence'))
    presence['from'] = 'bob@localhost/Bob'
    presence['to'] = 'test@localhost/Resource'
    c = presence.addElement('c')
    c['xmlns'] = 'http://jabber.org/protocol/caps'
    c['node'] = 'http://example.com/ICantBelieveItsNotTelepathy'
    c['ver'] = '1.2.3'
    stream.send(presence)

    event = q.expect('stream-iq', iq_type='get',
        query_ns='http://jabber.org/protocol/disco#info',
        to='bob@localhost/Bob')
    result = event.stanza
    result['type'] = 'result'
    assert event.query['node'] == \
        'http://example.com/ICantBelieveItsNotTelepathy#1.2.3'
    feature = event.query.addElement('feature')
    feature['var'] = ns.TUBES
    stream.send(result)
<commit_msg>accept-private-stream-tube: call sync_dbus to ensure that contact list channels have been announced<commit_after>"""
Receives several tube offers:
- Test to accept a 1-1 stream tube
  - using UNIX sockets and IPv4 sockets
  - using the old tube iface and the new tube iface
- Test to accept with bad parameters
- Test to refuse the tube offer
  - using the old tube iface and the new tube iface
"""

import dbus

from servicetest import call_async, EventPattern, EventProtocolClientFactory, sync_dbus
from gabbletest import acknowledge_iq

from twisted.words.xish import domish, xpath
from twisted.internet import reactor
import ns
import constants as cs
from bytestream import create_from_si_offer
import tubetestutil as t

bob_jid = 'bob@localhost/Bob'
stream_tube_id = 49

def receive_tube_offer(q, bus, conn, stream):
    message = domish.Element(('jabber:client', 'message'))
    message['to'] = 'test@localhost/Resource'
    message['from'] = bob_jid
    tube_node = message.addElement((ns.TUBES, 'tube'))
    tube_node['type'] = 'stream'
    tube_node['service'] = 'http'
    tube_node['id'] = str(stream_tube_id)
    stream.send(message)

    old_sig, new_sig = q.expect_many(
        EventPattern('dbus-signal', signal='NewChannel'),
        EventPattern('dbus-signal', signal='NewChannels'),
        )
    chan_path = old_sig.args[0]
    assert old_sig.args[1] == cs.CHANNEL_TYPE_TUBES, old_sig.args[1]
    assert old_sig.args[2] == cs.HT_CONTACT
    bob_handle = old_sig.args[3]
    assert old_sig.args[2] == 1, old_sig.args[2] # Suppress_Handler
    assert len(new_sig.args) == 1
    assert len(new_sig.args[0]) == 1
    assert new_sig.args[0][0][0] == chan_path, new_sig.args[0][0]
    assert new_sig.args[0][0][1] is not None

    event = q.expect('dbus-signal', signal='NewTube')

    old_sig, new_sig = q.expect_many(
        EventPattern('dbus-signal', signal='NewChannel'),
        EventPattern('dbus-signal', signal='NewChannels'),
        )
    new_chan_path = old_sig.args[0]
    assert new_chan_path != chan_path
    assert old_sig.args[1] == cs.CHANNEL_TYPE_STREAM_TUBE, old_sig.args[1]
    assert old_sig.args[2] == cs.HT_CONTACT
    bob_handle = old_sig.args[3]
    assert old_sig.args[2] == 1, old_sig.args[2] # Suppress_Handler
    assert len(new_sig.args) == 1
    assert len(new_sig.args[0]) == 1
    assert new_sig.args[0][0][0] == new_chan_path, new_sig.args[0][0]
    assert new_sig.args[0][0][1] is not None

    # create channel proxies
    tubes_chan = bus.get_object(conn.bus_name, chan_path)
    tubes_iface = dbus.Interface(tubes_chan, cs.CHANNEL_TYPE_TUBES)

    new_tube_chan = bus.get_object(conn.bus_name, new_chan_path)
    new_tube_iface = dbus.Interface(new_tube_chan, cs.CHANNEL_TYPE_STREAM_TUBE)

    return (tubes_chan, tubes_iface, new_tube_chan, new_tube_iface)

def expect_tube_activity(q, bus, conn, stream, bytestream_cls):
    event_socket, event_iq = q.expect_many(
            EventPattern('socket-connected'),
            EventPattern('stream-iq', to=bob_jid, query_ns=ns.SI,
                query_name='si'))

    protocol = event_socket.protocol
    protocol.sendData("hello initiator")

    bytestream, profile = create_from_si_offer(stream, q, bytestream_cls, event_iq.stanza,
        'test@localhost/Resource')

    assert profile == ns.TUBES

    stream_node = xpath.queryForNodes('/iq/si/stream[@xmlns="%s"]' %
        ns.TUBES, event_iq.stanza)[0]
    assert stream_node is not None
    assert stream_node['tube'] == str(stream_tube_id)

    result, si = bytestream.create_si_reply(event_iq.stanza)
    si.addElement((ns.TUBES, 'tube'))
    stream.send(result)

    bytestream.wait_bytestream_open()

    data = bytestream.get_data()
    assert data == 'hello initiator'

    # reply to the initiator
    bytestream.send_data('hello joiner')

    e = q.expect('socket-data')
    assert e.data == 'hello joiner'

    return bytestream

def test(q, bus, conn, stream, bytestream_cls):
    conn.Connect()

    _, vcard_event, roster_event = q.expect_many(
        EventPattern('dbus-signal', signal='StatusChanged', args=[0, 1]),
        EventPattern('stream-iq', to=None, query_ns='vcard-temp',
            query_name='vCard'),
        EventPattern('stream-iq', query_ns='jabber:iq:roster'))

    acknowledge_iq(stream, vcard_event.stanza)

    roster = roster_event.stanza
    roster['type'] = 'result'
    item = roster_event.query.addElement('item')
    item['jid'] = 'bob@localhost' # Bob can do tubes
    item['subscription'] = 'both'
    stream.send(roster)

    # Send Bob presence and his caps
    presence = domish.Element(('jabber:client', 'presence'))
    presence['from'] = 'bob@localhost/Bob'
    presence['to'] = 'test@localhost/Resource'
    c = presence.addElement('c')
    c['xmlns'] = 'http://jabber.org/protocol/caps'
    c['node'] = 'http://example.com/ICantBelieveItsNotTelepathy'
    c['ver'] = '1.2.3'
    stream.send(presence)

    event = q.expect('stream-iq', iq_type='get',
        query_ns='http://jabber.org/protocol/disco#info',
        to='bob@localhost/Bob')
    result = event.stanza
    result['type'] = 'result'
    assert event.query['node'] == \
        'http://example.com/ICantBelieveItsNotTelepathy#1.2.3'
    feature = event.query.addElement('feature')
    feature['var'] = ns.TUBES
    stream.send(result)

    sync_dbus(bus, q, conn)

{'fraction_non_alphanumeric': 0.13331438317225697, 'fraction_numerical': 0.009854083759711958, 'mean_word_length': 3.793369663941871, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, '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': '6607586', 'n_tokens_mistral': 3786, 'n_tokens_neox': 3635, 'n_words': 727}