<commit_before>"""
Test exposing incoming <hold/>, <ringing/> and <active/> notifications via the
CallState interface.
"""

from twisted.words.xish import xpath

from gabbletest import make_result_iq
from servicetest import (
    wrap_channel, make_channel_proxy, EventPattern, tp_path_prefix, sync_dbus)
import ns
import constants as cs

from jingletest2 import JingleTest2, test_all_dialects

def test(jp, q, bus, conn, stream):
    remote_jid = '<EMAIL>/Foo'
    jt = JingleTest2(jp, conn, q, stream, 'test@localhost', remote_jid)

    jt.prepare()

    self_handle = conn.GetSelfHandle()
    handle = conn.RequestHandles(cs.HT_CONTACT, [remote_jid])[0]

    path = conn.RequestChannel(cs.CHANNEL_TYPE_STREAMED_MEDIA, cs.HT_CONTACT,
        handle, True)

    chan = wrap_channel(bus.get_object(conn.bus_name, path), 'StreamedMedia',
        ['MediaSignalling', 'Group', 'CallState'])
    chan_props = chan.Properties.GetAll(cs.CHANNEL)
    assert cs.CHANNEL_IFACE_CALL_STATE in chan_props['Interfaces'], \
        chan_props['Interfaces']

    call_states = chan.CallState.GetCallStates()
    assert call_states == { handle: 0 } or \
        call_states == {}, call_states

    chan.StreamedMedia.RequestStreams(handle, [cs.MEDIA_STREAM_TYPE_AUDIO])

    # S-E gets notified about new session handler, and calls Ready on it
    e = q.expect('dbus-signal', signal='NewSessionHandler')
    assert e.args[1] == 'rtp'

    session_handler = make_channel_proxy(conn, e.args[0], 'Media.SessionHandler')
    session_handler.Ready()

    e = q.expect('dbus-signal', signal='NewStreamHandler')
    audio_path = e.args[0]
    audio_path_suffix = audio_path[len(tp_path_prefix):]
    stream_handler = make_channel_proxy(conn, audio_path, 'Media.StreamHandler')

    stream_handler.NewNativeCandidate("fake", jt.get_remote_transports_dbus())
    stream_handler.Ready(jt.get_audio_codecs_dbus())
    stream_handler.StreamState(cs.MEDIA_STREAM_STATE_CONNECTED)

    e = q.expect('stream-iq', predicate=jp.action_predicate('session-initiate'))
    stream.send(make_result_iq(stream, e.stanza))

    jt.parse_session_initiate(e.query)

    if jp.is_modern_jingle():
        # The other person's client starts ringing, and tells us so!
        node = jp.SetIq(jt.peer, jt.jid, [
            jp.Jingle(jt.sid, jt.jid, 'session-info', [
                ('ringing', ns.JINGLE_RTP_INFO_1, {}, []) ]) ])
        stream.send(jp.xml(node))

    # If this is an old Jingle dialect, Gabble should treat the
    # session-initiate ack as ringing notification; if it's modern Jingle, we
    # just sent a ringing notification.
    q.expect('dbus-signal', signal='CallStateChanged',
            args=[handle, cs.CALL_STATE_RINGING])

    call_states = chan.CallState.GetCallStates()
    assert call_states == { handle: cs.CALL_STATE_RINGING }, call_states

    if jp.is_modern_jingle():
        # We're waiting in a queue, so the other person's client tells us we're on
        # hold. Gabble should ack the IQ, and set the call state to Ringing | Held.
        # Also, Gabble certainly shouldn't tell s-e to start sending. (Although it
        # might tell it not to; we don't mind.)
        node = jp.SetIq(jt.peer, jt.jid, [
            jp.Jingle(jt.sid, jt.jid, 'session-info', [
                ('hold', ns.JINGLE_RTP_INFO_1, {}, []) ]) ])
        stream.send(jp.xml(node))

        forbidden = [
            EventPattern('dbus-signal', signal='SetStreamSending', args=[True],
                path=audio_path_suffix),
                ]
        q.forbid_events(forbidden)

        q.expect_many(
            EventPattern('stream-iq', iq_type='result', iq_id=node[2]['id']),
            EventPattern('dbus-signal', signal='CallStateChanged',
                args=[handle, cs.CALL_STATE_RINGING | cs.CALL_STATE_HELD]),
            )

        call_states = chan.CallState.GetCallStates()
        assert call_states == { handle: cs.CALL_STATE_RINGING | cs.CALL_STATE_HELD }, call_states

        # We're at the head of a queue, so the other person's client tells us we're
        # no longer on hold. The call centre phone's still ringing, though. s-e
        # still shouldn't start sending.
        node = jp.SetIq(jt.peer, jt.jid, [
            jp.Jingle(jt.sid, jt.jid, 'session-info', [
                ('unhold', ns.JINGLE_RTP_INFO_1, {}, []) ]) ])
        stream.send(jp.xml(node))

        q.expect_many(
            EventPattern('stream-iq', iq_type='result', iq_id=node[2]['id']),
            EventPattern('dbus-signal', signal='CallStateChanged',
                args=[handle, cs.CALL_STATE_RINGING]),
            )

        call_states = chan.CallState.GetCallStates()
        assert call_states == { handle: cs.CALL_STATE_RINGING }, call_states

        sync_dbus(bus, q, conn)
        q.unforbid_events(forbidden)

    jt.accept()

    if jp.is_modern_jingle():
        # The other person's client decides it's not ringing any more
        node = jp.SetIq(jt.peer, jt.jid, [
            jp.Jingle(jt.sid, jt.jid, 'session-info', [
                ('active', ns.JINGLE_RTP_INFO_1, {}, []) ]) ])
        stream.send(jp.xml(node))

    # Gabble tells s-e to start sending, and removes the Ringing flag, either
    # because we got <active/> or because of the session-accept in ye olde
    # Jingle.
    q.expect_many(
        EventPattern('dbus-signal', signal='SetStreamSending', args=[True],
            path=audio_path_suffix),
        EventPattern('dbus-signal', signal='CallStateChanged',
            args=[ handle, 0 ]),
        )

    call_states = chan.CallState.GetCallStates()
    assert call_states == { handle: 0 } or call_states == {}, call_states

    # The rest of the test concerns things we only support in the glorious
    # modern Jingle future.
    if not jp.is_modern_jingle():
        conn.Disconnect()
        return

    # The other person puts us on hold.  Gabble should ack the session-info IQ,
    # tell s-e to stop sending on the held stream, and set the call state.
    node = jp.SetIq(jt.peer, jt.jid, [
        jp.Jingle(jt.sid, jt.jid, 'session-info', [
            ('hold', ns.JINGLE_RTP_INFO_1, {}, []) ]) ])
    stream.send(jp.xml(node))

    q.expect_many(<commit_msg>jingle/call-state.py: there it no point to disconnect the connection when we short-circuits the test<commit_after>"""
Test exposing incoming <hold/>, <ringing/> and <active/> notifications via the
CallState interface.
"""

from twisted.words.xish import xpath

from gabbletest import make_result_iq
from servicetest import (
    wrap_channel, make_channel_proxy, EventPattern, tp_path_prefix, sync_dbus)
import ns
import constants as cs

from jingletest2 import JingleTest2, test_all_dialects

def test(jp, q, bus, conn, stream):
    remote_jid = '<EMAIL>/Foo'
    jt = JingleTest2(jp, conn, q, stream, 'test@localhost', remote_jid)

    jt.prepare()

    self_handle = conn.GetSelfHandle()
    handle = conn.RequestHandles(cs.HT_CONTACT, [remote_jid])[0]

    path = conn.RequestChannel(cs.CHANNEL_TYPE_STREAMED_MEDIA, cs.HT_CONTACT,
        handle, True)

    chan = wrap_channel(bus.get_object(conn.bus_name, path), 'StreamedMedia',
        ['MediaSignalling', 'Group', 'CallState'])
    chan_props = chan.Properties.GetAll(cs.CHANNEL)
    assert cs.CHANNEL_IFACE_CALL_STATE in chan_props['Interfaces'], \
        chan_props['Interfaces']

    call_states = chan.CallState.GetCallStates()
    assert call_states == { handle: 0 } or \
        call_states == {}, call_states

    chan.StreamedMedia.RequestStreams(handle, [cs.MEDIA_STREAM_TYPE_AUDIO])

    # S-E gets notified about new session handler, and calls Ready on it
    e = q.expect('dbus-signal', signal='NewSessionHandler')
    assert e.args[1] == 'rtp'

    session_handler = make_channel_proxy(conn, e.args[0], 'Media.SessionHandler')
    session_handler.Ready()

    e = q.expect('dbus-signal', signal='NewStreamHandler')
    audio_path = e.args[0]
    audio_path_suffix = audio_path[len(tp_path_prefix):]
    stream_handler = make_channel_proxy(conn, audio_path, 'Media.StreamHandler')

    stream_handler.NewNativeCandidate("fake", jt.get_remote_transports_dbus())
    stream_handler.Ready(jt.get_audio_codecs_dbus())
    stream_handler.StreamState(cs.MEDIA_STREAM_STATE_CONNECTED)

    e = q.expect('stream-iq', predicate=jp.action_predicate('session-initiate'))
    stream.send(make_result_iq(stream, e.stanza))

    jt.parse_session_initiate(e.query)

    if jp.is_modern_jingle():
        # The other person's client starts ringing, and tells us so!
        node = jp.SetIq(jt.peer, jt.jid, [
            jp.Jingle(jt.sid, jt.jid, 'session-info', [
                ('ringing', ns.JINGLE_RTP_INFO_1, {}, []) ]) ])
        stream.send(jp.xml(node))

    # If this is an old Jingle dialect, Gabble should treat the
    # session-initiate ack as ringing notification; if it's modern Jingle, we
    # just sent a ringing notification.
    q.expect('dbus-signal', signal='CallStateChanged',
            args=[handle, cs.CALL_STATE_RINGING])

    call_states = chan.CallState.GetCallStates()
    assert call_states == { handle: cs.CALL_STATE_RINGING }, call_states

    if jp.is_modern_jingle():
        # We're waiting in a queue, so the other person's client tells us we're on
        # hold. Gabble should ack the IQ, and set the call state to Ringing | Held.
        # Also, Gabble certainly shouldn't tell s-e to start sending. (Although it
        # might tell it not to; we don't mind.)
        node = jp.SetIq(jt.peer, jt.jid, [
            jp.Jingle(jt.sid, jt.jid, 'session-info', [
                ('hold', ns.JINGLE_RTP_INFO_1, {}, []) ]) ])
        stream.send(jp.xml(node))

        forbidden = [
            EventPattern('dbus-signal', signal='SetStreamSending', args=[True],
                path=audio_path_suffix),
                ]
        q.forbid_events(forbidden)

        q.expect_many(
            EventPattern('stream-iq', iq_type='result', iq_id=node[2]['id']),
            EventPattern('dbus-signal', signal='CallStateChanged',
                args=[handle, cs.CALL_STATE_RINGING | cs.CALL_STATE_HELD]),
            )

        call_states = chan.CallState.GetCallStates()
        assert call_states == { handle: cs.CALL_STATE_RINGING | cs.CALL_STATE_HELD }, call_states

        # We're at the head of a queue, so the other person's client tells us we're
        # no longer on hold. The call centre phone's still ringing, though. s-e
        # still shouldn't start sending.
        node = jp.SetIq(jt.peer, jt.jid, [
            jp.Jingle(jt.sid, jt.jid, 'session-info', [
                ('unhold', ns.JINGLE_RTP_INFO_1, {}, []) ]) ])
        stream.send(jp.xml(node))

        q.expect_many(
            EventPattern('stream-iq', iq_type='result', iq_id=node[2]['id']),
            EventPattern('dbus-signal', signal='CallStateChanged',
                args=[handle, cs.CALL_STATE_RINGING]),
            )

        call_states = chan.CallState.GetCallStates()
        assert call_states == { handle: cs.CALL_STATE_RINGING }, call_states

        sync_dbus(bus, q, conn)
        q.unforbid_events(forbidden)

    jt.accept()

    if jp.is_modern_jingle():
        # The other person's client decides it's not ringing any more
        node = jp.SetIq(jt.peer, jt.jid, [
            jp.Jingle(jt.sid, jt.jid, 'session-info', [
                ('active', ns.JINGLE_RTP_INFO_1, {}, []) ]) ])
        stream.send(jp.xml(node))

    # Gabble tells s-e to start sending, and removes the Ringing flag, either
    # because we got <active/> or because of the session-accept in ye olde
    # Jingle.
    q.expect_many(
        EventPattern('dbus-signal', signal='SetStreamSending', args=[True],
            path=audio_path_suffix),
        EventPattern('dbus-signal', signal='CallStateChanged',
            args=[ handle, 0 ]),
        )

    call_states = chan.CallState.GetCallStates()
    assert call_states == { handle: 0 } or call_states == {}, call_states

    # The rest of the test concerns things we only support in the glorious
    # modern Jingle future.
    if not jp.is_modern_jingle():
        return

    # The other person puts us on hold.  Gabble should ack the session-info IQ,
    # tell s-e to stop sending on the held stream, and set the call state.
    node = jp.SetIq(jt.peer, jt.jid, [
        jp.Jingle(jt.sid, jt.jid, 'session-info', [
            ('hold', ns.JINGLE_RTP_INFO_1, {}, []) ]) ])
    stream.send(jp.xml(node))

    q.expect_many(
{'fraction_non_alphanumeric': 0.12247850196897854, 'fraction_numerical': 0.002732460017680624, 'mean_word_length': 2.9529860228716647, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 10}, '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': '6325891', 'n_tokens_mistral': 4268, 'n_tokens_neox': 4032, 'n_words': 925}