<commit_before>"""
Test basic outgoing call handling, using CreateChannel and all three variations
of RequestChannel.
"""
from functools import partial
import dbus
from twisted.words.xish import xpath

from gabbletest import exec_test
from servicetest import (
    make_channel_proxy, wrap_channel,
    EventPattern, call_async,
    assertEquals, assertContains, assertLength,
    )
import constants as cs
from jingletest2 import JingleTest2, test_all_dialects

# There are various deprecated APIs for requesting calls, documented at
# <http://telepathy.freedesktop.org/wiki/Requesting StreamedMedia channels>.
# These are ordered from most recent to most deprecated.
CREATE = 0 # CreateChannel({TargetHandleType: Contact, TargetHandle: h});
           # RequestStreams()
REQUEST_ANONYMOUS = 1 # RequestChannel(HandleTypeNone, 0); RequestStreams()
REQUEST_ANONYMOUS_AND_ADD = 2 # RequestChannel(HandleTypeNone, 0);
                              # AddMembers([h], ...); RequestStreams(h,...)
REQUEST_NONYMOUS = 3 # RequestChannel(HandleTypeContact, h);
                     # RequestStreams(h, ...)

def create(jp, q, bus, conn, stream, peer='foo<EMAIL>.com/Res'):
    worker(jp, q, bus, conn, stream, CREATE, peer)

def request_anonymous(jp, q, bus, conn, stream, peer='publish@foo.com/Res'):
    worker(jp, q, bus, conn, stream, REQUEST_ANONYMOUS, peer)

def request_anonymous_and_add(jp, q, bus, conn, stream,
        peer='publish-subscribe@foo.com/Res'):
    worker(jp, q, bus, conn, stream, REQUEST_ANONYMOUS_AND_ADD, peer)

def request_nonymous(jp, q, bus, conn, stream, peer='subscribe@foo.com/Res'):
    worker(jp, q, bus, conn, stream, REQUEST_NONYMOUS, peer)

def worker(jp, q, bus, conn, stream, variant, peer):
    jt2 = JingleTest2(jp, conn, q, stream, 'test@localhost', peer)
    jt2.prepare(send_presence=True, send_roster=True)

    self_handle = conn.GetSelfHandle()
    remote_handle = conn.RequestHandles(1, [jt2.peer])[0]

    if variant == REQUEST_NONYMOUS:
        path = conn.RequestChannel(cs.CHANNEL_TYPE_STREAMED_MEDIA,
            cs.HT_CONTACT, remote_handle, True)
    elif variant == CREATE:
        path = conn.Requests.CreateChannel({
            cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_STREAMED_MEDIA,
            cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
            cs.TARGET_HANDLE: remote_handle,
            })[0]
    else:
        path = conn.RequestChannel(cs.CHANNEL_TYPE_STREAMED_MEDIA,
            cs.HT_NONE, 0, True)

    old_sig, new_sig = q.expect_many(
        EventPattern('dbus-signal', signal='NewChannel',
            predicate=lambda e: cs.CHANNEL_TYPE_CONTACT_LIST not in e.args),
        EventPattern('dbus-signal', signal='NewChannels',
            predicate=lambda e:
                cs.CHANNEL_TYPE_CONTACT_LIST not in e.args[0][0][1].values()),
        )

    if variant == REQUEST_NONYMOUS or variant == CREATE:
        assertEquals( [path, cs.CHANNEL_TYPE_STREAMED_MEDIA, cs.HT_CONTACT,
            remote_handle, True], old_sig.args)
    else:
        assertEquals( [path, cs.CHANNEL_TYPE_STREAMED_MEDIA, cs.HT_NONE, 0,
            True], old_sig.args)

    assertLength(1, new_sig.args)
    assertLength(1, new_sig.args[0])       # one channel
    assertLength(2, new_sig.args[0][0])    # two struct members
    emitted_props = new_sig.args[0][0][1]

    assertEquals(
        cs.CHANNEL_TYPE_STREAMED_MEDIA, emitted_props[cs.CHANNEL_TYPE])

    if variant == REQUEST_NONYMOUS or variant == CREATE:
        assertEquals(remote_handle, emitted_props[cs.TARGET_HANDLE])
        assertEquals(cs.HT_CONTACT, emitted_props[cs.TARGET_HANDLE_TYPE])
        assertEquals(jt2.peer_bare_jid, emitted_props[cs.TARGET_ID])
    else:
        assertEquals(0, emitted_props[cs.TARGET_HANDLE])
        assertEquals(cs.HT_NONE, emitted_props[cs.TARGET_HANDLE_TYPE])
        assertEquals('', emitted_props[cs.TARGET_ID])

    assertEquals(True, emitted_props[cs.REQUESTED])
    assertEquals(self_handle, emitted_props[cs.INITIATOR_HANDLE])
    assertEquals('test@localhost', emitted_props[cs.INITIATOR_ID])

    chan = wrap_channel(bus.get_object(conn.bus_name, path), 'StreamedMedia',
        ['MediaSignalling'])

    # Exercise basic Channel Properties
    channel_props = chan.Properties.GetAll(cs.CHANNEL)

    assertEquals(cs.CHANNEL_TYPE_STREAMED_MEDIA,
        channel_props.get('ChannelType'))

    if variant == REQUEST_NONYMOUS or variant == CREATE:
        assertEquals(remote_handle, channel_props['TargetHandle'])
        assertEquals(cs.HT_CONTACT, channel_props['TargetHandleType'])
        assertEquals(jt2.peer_bare_jid, channel_props['TargetID'])
        assertEquals((cs.HT_CONTACT, remote_handle), chan.GetHandle())
    else:
        assertEquals(0, channel_props['TargetHandle'])
        assertEquals(cs.HT_NONE, channel_props['TargetHandleType'])
        assertEquals('', channel_props['TargetID'])
        assertEquals((cs.HT_NONE, 0), chan.GetHandle())

    for interface in [
            cs.CHANNEL_IFACE_GROUP, cs.CHANNEL_IFACE_MEDIA_SIGNALLING,
            cs.TP_AWKWARD_PROPERTIES, cs.CHANNEL_IFACE_HOLD]:
        assertContains(interface, channel_props['Interfaces'])

    assertEquals(True, channel_props['Requested'])
    assertEquals('test@localhost', channel_props['InitiatorID'])
    assertEquals(conn.GetSelfHandle(), channel_props['InitiatorHandle'])

    # Exercise Group Properties
    group_props = chan.Properties.GetAll(cs.CHANNEL_IFACE_GROUP)

    assertEquals([self_handle], group_props['Members'])
    assertEquals([], group_props['LocalPendingMembers'])

    if variant == REQUEST_NONYMOUS:
        # In this variant, they're meant to be in RP even though we've sent
        # nothing
        assertEquals([remote_handle], group_props['RemotePendingMembers'])
    else:
        # For an anonymous channel, the peer isn't yet known; for a Create-d
        # channel, the peer only appears in RP when we actually send them the
        # session-initiate
        assertEquals([], group_props['RemotePendingMembers'])

        if variant == REQUEST_ANONYMOUS_AND_ADD:
            # but we should be allowed to add the peer.
            chan.Group.AddMembers([remote_handle], 'I love backwards compat')

    base_flags = cs.GF_PROPERTIES | cs.GF_MESSAGE_REMOVE \
               | cs.GF_MESSAGE_REJECT | cs.GF_MESSAGE_RESCIND

    if variant == REQUEST_ANONYMOUS_AND_ADD or variant == REQUEST_ANONYMOUS:
        expected_flags = base_flags | cs.GF_CAN_ADD
    else:
        expected_flags = base_flags
    assertEquals(expected_flags, group_props['GroupFlags'])
    assertEquals({}, group_props['HandleOwners'])

    assertEquals([], chan.StreamedMedia.ListStreams())
    streams = chan.StreamedMedia.RequestStreams(remote_handle,
        [cs.MEDIA_STREAM_TYPE_AUDIO])<commit_msg>jingle outgoing test: add MembersChangedDetailed flag to base flags<commit_after>"""
Test basic outgoing call handling, using CreateChannel and all three variations
of RequestChannel.
"""
from functools import partial
import dbus
from twisted.words.xish import xpath

from gabbletest import exec_test
from servicetest import (
    make_channel_proxy, wrap_channel,
    EventPattern, call_async,
    assertEquals, assertContains, assertLength,
    )
import constants as cs
from jingletest2 import JingleTest2, test_all_dialects

# There are various deprecated APIs for requesting calls, documented at
# <http://telepathy.freedesktop.org/wiki/Requesting StreamedMedia channels>.
# These are ordered from most recent to most deprecated.
CREATE = 0 # CreateChannel({TargetHandleType: Contact, TargetHandle: h});
           # RequestStreams()
REQUEST_ANONYMOUS = 1 # RequestChannel(HandleTypeNone, 0); RequestStreams()
REQUEST_ANONYMOUS_AND_ADD = 2 # RequestChannel(HandleTypeNone, 0);
                              # AddMembers([h], ...); RequestStreams(h,...)
REQUEST_NONYMOUS = 3 # RequestChannel(HandleTypeContact, h);
                     # RequestStreams(h, ...)

def create(jp, q, bus, conn, stream, peer='<EMAIL>/Res'):
    worker(jp, q, bus, conn, stream, CREATE, peer)

def request_anonymous(jp, q, bus, conn, stream, peer='<EMAIL>/Res'):
    worker(jp, q, bus, conn, stream, REQUEST_ANONYMOUS, peer)

def request_anonymous_and_add(jp, q, bus, conn, stream,
        peer='publish-subscribe@foo.com/Res'):
    worker(jp, q, bus, conn, stream, REQUEST_ANONYMOUS_AND_ADD, peer)

def request_nonymous(jp, q, bus, conn, stream, peer='subscribe@foo.com/Res'):
    worker(jp, q, bus, conn, stream, REQUEST_NONYMOUS, peer)

def worker(jp, q, bus, conn, stream, variant, peer):
    jt2 = JingleTest2(jp, conn, q, stream, 'test@localhost', peer)
    jt2.prepare(send_presence=True, send_roster=True)

    self_handle = conn.GetSelfHandle()
    remote_handle = conn.RequestHandles(1, [jt2.peer])[0]

    if variant == REQUEST_NONYMOUS:
        path = conn.RequestChannel(cs.CHANNEL_TYPE_STREAMED_MEDIA,
            cs.HT_CONTACT, remote_handle, True)
    elif variant == CREATE:
        path = conn.Requests.CreateChannel({
            cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_STREAMED_MEDIA,
            cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
            cs.TARGET_HANDLE: remote_handle,
            })[0]
    else:
        path = conn.RequestChannel(cs.CHANNEL_TYPE_STREAMED_MEDIA,
            cs.HT_NONE, 0, True)

    old_sig, new_sig = q.expect_many(
        EventPattern('dbus-signal', signal='NewChannel',
            predicate=lambda e: cs.CHANNEL_TYPE_CONTACT_LIST not in e.args),
        EventPattern('dbus-signal', signal='NewChannels',
            predicate=lambda e:
                cs.CHANNEL_TYPE_CONTACT_LIST not in e.args[0][0][1].values()),
        )

    if variant == REQUEST_NONYMOUS or variant == CREATE:
        assertEquals( [path, cs.CHANNEL_TYPE_STREAMED_MEDIA, cs.HT_CONTACT,
            remote_handle, True], old_sig.args)
    else:
        assertEquals( [path, cs.CHANNEL_TYPE_STREAMED_MEDIA, cs.HT_NONE, 0,
            True], old_sig.args)

    assertLength(1, new_sig.args)
    assertLength(1, new_sig.args[0])       # one channel
    assertLength(2, new_sig.args[0][0])    # two struct members
    emitted_props = new_sig.args[0][0][1]

    assertEquals(
        cs.CHANNEL_TYPE_STREAMED_MEDIA, emitted_props[cs.CHANNEL_TYPE])

    if variant == REQUEST_NONYMOUS or variant == CREATE:
        assertEquals(remote_handle, emitted_props[cs.TARGET_HANDLE])
        assertEquals(cs.HT_CONTACT, emitted_props[cs.TARGET_HANDLE_TYPE])
        assertEquals(jt2.peer_bare_jid, emitted_props[cs.TARGET_ID])
    else:
        assertEquals(0, emitted_props[cs.TARGET_HANDLE])
        assertEquals(cs.HT_NONE, emitted_props[cs.TARGET_HANDLE_TYPE])
        assertEquals('', emitted_props[cs.TARGET_ID])

    assertEquals(True, emitted_props[cs.REQUESTED])
    assertEquals(self_handle, emitted_props[cs.INITIATOR_HANDLE])
    assertEquals('test@localhost', emitted_props[cs.INITIATOR_ID])

    chan = wrap_channel(bus.get_object(conn.bus_name, path), 'StreamedMedia',
        ['MediaSignalling'])

    # Exercise basic Channel Properties
    channel_props = chan.Properties.GetAll(cs.CHANNEL)

    assertEquals(cs.CHANNEL_TYPE_STREAMED_MEDIA,
        channel_props.get('ChannelType'))

    if variant == REQUEST_NONYMOUS or variant == CREATE:
        assertEquals(remote_handle, channel_props['TargetHandle'])
        assertEquals(cs.HT_CONTACT, channel_props['TargetHandleType'])
        assertEquals(jt2.peer_bare_jid, channel_props['TargetID'])
        assertEquals((cs.HT_CONTACT, remote_handle), chan.GetHandle())
    else:
        assertEquals(0, channel_props['TargetHandle'])
        assertEquals(cs.HT_NONE, channel_props['TargetHandleType'])
        assertEquals('', channel_props['TargetID'])
        assertEquals((cs.HT_NONE, 0), chan.GetHandle())

    for interface in [
            cs.CHANNEL_IFACE_GROUP, cs.CHANNEL_IFACE_MEDIA_SIGNALLING,
            cs.TP_AWKWARD_PROPERTIES, cs.CHANNEL_IFACE_HOLD]:
        assertContains(interface, channel_props['Interfaces'])

    assertEquals(True, channel_props['Requested'])
    assertEquals('test@localhost', channel_props['InitiatorID'])
    assertEquals(conn.GetSelfHandle(), channel_props['InitiatorHandle'])

    # Exercise Group Properties
    group_props = chan.Properties.GetAll(cs.CHANNEL_IFACE_GROUP)

    assertEquals([self_handle], group_props['Members'])
    assertEquals([], group_props['LocalPendingMembers'])

    if variant == REQUEST_NONYMOUS:
        # In this variant, they're meant to be in RP even though we've sent
        # nothing
        assertEquals([remote_handle], group_props['RemotePendingMembers'])
    else:
        # For an anonymous channel, the peer isn't yet known; for a Create-d
        # channel, the peer only appears in RP when we actually send them the
        # session-initiate
        assertEquals([], group_props['RemotePendingMembers'])

        if variant == REQUEST_ANONYMOUS_AND_ADD:
            # but we should be allowed to add the peer.
            chan.Group.AddMembers([remote_handle], 'I love backwards compat')

    base_flags = cs.GF_MEMBERS_CHANGED_DETAILED | cs.GF_PROPERTIES | cs.GF_MESSAGE_REMOVE \
               | cs.GF_MESSAGE_REJECT | cs.GF_MESSAGE_RESCIND

    if variant == REQUEST_ANONYMOUS_AND_ADD or variant == REQUEST_ANONYMOUS:
        expected_flags = base_flags | cs.GF_CAN_ADD
    else:
        expected_flags = base_flags
    assertEquals(expected_flags, group_props['GroupFlags'])
    assertEquals({}, group_props['HandleOwners'])

    assertEquals([], chan.StreamedMedia.ListStreams())
    streams = chan.StreamedMedia.RequestStreams(remote_handle,
        [cs.MEDIA_STREAM_TYPE_AUDIO])
{'fraction_non_alphanumeric': 0.10662739322533137, 'fraction_numerical': 0.005007363770250369, 'mean_word_length': 3.785412262156448, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 5, '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': '4726085', 'n_tokens_mistral': 4340, 'n_tokens_neox': 4086, 'n_words': 868}