/usr/lib/telepathy-gabble-tests/twisted/jingle/test-incoming-call-reject.py is in telepathy-gabble-tests 0.18.4-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | """
Test incoming call handling - reject a call because we're busy, and for no
reason.
"""
from twisted.words.xish import xpath
from servicetest import make_channel_proxy, EventPattern, call_async
from jingletest2 import JingleTest2, test_all_dialects
import constants as cs
from config import VOIP_ENABLED
if not VOIP_ENABLED:
print "NOTE: built with --disable-voip"
raise SystemExit(77)
def test_busy(jp, q, bus, conn, stream):
test(jp, q, bus, conn, stream, True)
def test_no_reason(jp, q, bus, conn, stream):
test(jp, q, bus, conn, stream, False)
def test(jp, q, bus, conn, stream, busy):
remote_jid = 'foo@bar.com/Foo'
jt = JingleTest2(jp, conn, q, stream, 'test@localhost', remote_jid)
jt.prepare()
self_handle = conn.GetSelfHandle()
remote_handle = conn.RequestHandles(cs.HT_CONTACT, [remote_jid])[0]
# Remote end calls us
jt.incoming_call()
# FIXME: these signals are not observable by real clients, since they
# happen before NewChannels.
# The caller is in members
e = q.expect('dbus-signal', signal='MembersChanged',
args=[u'', [remote_handle], [], [], [], 0, 0])
# We're pending because of remote_handle
e = q.expect('dbus-signal', signal='MembersChanged',
args=[u'', [], [], [self_handle], [], remote_handle,
cs.GC_REASON_INVITED])
# 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()
media_chan = make_channel_proxy(conn, e.path, 'Channel.Interface.Group')
# Exercise channel properties
channel_props = media_chan.GetAll(cs.CHANNEL,
dbus_interface=cs.PROPERTIES_IFACE)
assert channel_props['TargetHandle'] == remote_handle
assert channel_props['TargetHandleType'] == 1
assert channel_props['TargetID'] == 'foo@bar.com'
assert channel_props['Requested'] == False
assert channel_props['InitiatorID'] == 'foo@bar.com'
assert channel_props['InitiatorHandle'] == remote_handle
if busy:
# First, try using a reason that doesn't make any sense
call_async(q, media_chan, 'RemoveMembersWithReason',
[self_handle], "what kind of a reason is Separated?!",
cs.GC_REASON_SEPARATED)
e = q.expect('dbus-error', method='RemoveMembersWithReason')
assert e.error.get_dbus_name() == cs.INVALID_ARGUMENT
# Now try a more sensible reason.
media_chan.RemoveMembersWithReason([self_handle],
"which part of 'Do Not Disturb' don't you understand?",
cs.GC_REASON_BUSY)
else:
media_chan.RemoveMembers([self_handle], 'rejected')
iq, mc, _ = q.expect_many(
EventPattern('stream-iq',
predicate=jp.action_predicate('session-terminate')),
EventPattern('dbus-signal', signal='MembersChanged'),
EventPattern('dbus-signal', signal='Closed'),
)
_, added, removed, lp, rp, actor, reason = mc.args
assert added == [], added
assert set(removed) == set([self_handle, remote_handle]), \
(removed, self_handle, remote_handle)
assert lp == [], lp
assert rp == [], rp
assert actor == self_handle, (actor, self_handle)
if busy:
assert reason == cs.GC_REASON_BUSY, reason
else:
assert reason == cs.GC_REASON_NONE, reason
if jp.is_modern_jingle():
jingle = iq.query
if busy:
r = "/jingle/reason/busy"
else:
r = "/jingle/reason/cancel"
assert xpath.queryForNodes(r, jingle) is not None, (jingle.toXml(), r)
if __name__ == '__main__':
test_all_dialects(test_busy)
test_all_dialects(test_no_reason)
|