本文整理汇总了Python中servicetest.assertEquals函数的典型用法代码示例。如果您正苦于以下问题:Python assertEquals函数的具体用法?Python assertEquals怎么用?Python assertEquals使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了assertEquals函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: check_room_props
def check_room_props(chan, subject_str, actor):
# Room props
props = chan.GetAll(cs.CHANNEL_IFACE_ROOM,
dbus_interface=dbus.PROPERTIES_IFACE)
subject = props['Subject']
assertEquals(subject_str, subject[0])
assertEquals(actor, subject[1])
开发者ID:jku,项目名称:telepathy-gabble,代码行数:7,代码来源:subject.py
示例2: check_normal_props
def check_normal_props(props):
assertEquals(cs.CONTACT_INFO_FLAG_CAN_SET, props['ContactInfoFlags'])
sf = props['SupportedFields']
sf.sort()
for f in sf:
f[1].sort() # type-parameters
assertEquals([
('adr', types('home work postal parcel dom intl pref'), 0, UNLIMITED),
('bday', [], PARAMS_EXACT, UNLIMITED),
('email', types('home work internet pref x400'), 0, UNLIMITED),
('fn', [], PARAMS_EXACT, 1),
('geo', [], PARAMS_EXACT, 1),
('label', types('home work postal parcel dom intl pref'), 0,
UNLIMITED),
('mailer', [], PARAMS_EXACT, UNLIMITED),
('n', [], PARAMS_EXACT, 1),
('nickname', [], PARAMS_EXACT | OVERWRITTEN_BY_NICKNAME, UNLIMITED),
('note', [], PARAMS_EXACT, UNLIMITED),
('org', [], PARAMS_EXACT, UNLIMITED),
('prodid', [], PARAMS_EXACT, UNLIMITED),
('rev', [], PARAMS_EXACT, UNLIMITED),
('role', [], PARAMS_EXACT, UNLIMITED),
('sort-string', [], PARAMS_EXACT, UNLIMITED),
('tel', types('home work voice fax pager msg cell video bbs '
'modem isdn pcs pref'), 0, UNLIMITED),
('title', [], PARAMS_EXACT, UNLIMITED),
('tz', [], PARAMS_EXACT, UNLIMITED),
('uid', [], PARAMS_EXACT, UNLIMITED),
('url', [], PARAMS_EXACT, UNLIMITED),
('x-desc', [], PARAMS_EXACT, UNLIMITED),
('x-jabber', [], PARAMS_EXACT, UNLIMITED),
], sf)
开发者ID:Thaodan,项目名称:telepathy-gabble,代码行数:32,代码来源:supported-fields.py
示例3: _check_oob_iq
def _check_oob_iq(self, iq_event):
assert iq_event.iq_type == 'set'
assert iq_event.connection == self.incoming
self.iq = iq_event.stanza
assert self.iq['to'] == self.contact_name
query = self.iq.firstChildElement()
assert query.uri == 'jabber:iq:oob'
url_node = xpath.queryForNodes("/iq/query/url", self.iq)[0]
assert url_node['type'] == 'file'
assert url_node['size'] == str(self.file.size)
assert url_node['mimeType'] == self.file.content_type
self.url = url_node.children[0]
_, self.host, self.filename, _, _, _ = urlparse.urlparse(self.url)
urllib.unquote(self.filename) == self.file.name
desc_node = xpath.queryForNodes("/iq/query/desc", self.iq)[0]
self.desc = desc_node.children[0]
assert self.desc == self.file.description
# Metadata forms
forms = extract_data_forms(xpath.queryForNodes('/iq/query/x', self.iq))
if self.service_name:
assertEquals({'ServiceName': [self.service_name]},
forms[ns.TP_FT_METADATA_SERVICE])
else:
assert ns.TP_FT_METADATA_SERVICE not in forms
if self.metadata:
assertEquals(self.metadata, forms[ns.TP_FT_METADATA])
else:
assert ns.TP_FT_METADATA not in forms
开发者ID:freedesktop-unofficial-mirror,项目名称:telepathy__telepathy-salut,代码行数:31,代码来源:file_transfer_helper.py
示例4: test_privacy_list_push_conflict
def test_privacy_list_push_conflict(q, bus, conn, stream):
test_invisible_on_connect(q, bus, conn, stream)
set_id = stream.send_privacy_list_push_iq("invisible")
_, req_list = q.expect_many(
EventPattern('stream-iq', iq_type='result', predicate=lambda event: \
event.stanza['id'] == set_id),
EventPattern('stream-iq', query_ns=ns.PRIVACY, iq_type="get"))
stream.send_privacy_list(req_list.stanza,
[elem('item', type='jid', value='[email protected]', action='allow',
order='1')(elem('presence-out')),
elem('item', action='deny', order='2')(elem('presence-out'))])
create_list = q.expect('stream-iq', query_ns=ns.PRIVACY, iq_type='set')
created = xpath.queryForNodes('//list', create_list.stanza)[0]
assertEquals(created["name"], 'invisible-gabble')
acknowledge_iq(stream, create_list.stanza)
set_active = q.expect('stream-iq', query_ns=ns.PRIVACY, iq_type='set')
active = xpath.queryForNodes('//active', set_active.query)[0]
assertEquals('invisible-gabble', active['name'])
acknowledge_iq(stream, set_active.stanza)
开发者ID:jku,项目名称:telepathy-gabble,代码行数:25,代码来源:invisible_xep_0126.py
示例5: test_shared_status_list
def test_shared_status_list(q, bus, conn, stream):
'''Test the shared status list usage'''
test_statuses = {"dnd" : ['I am not available now',
'I am busy with real work',
'I have a life, you know...',
'I am actually playing Duke Nukem',
'It is important to me'],
"available" : ['I am twiddling my thumbs',
'Please chat me up',
'I am here for you',
'Message me already!']}
max_statuses = int(stream.max_statuses)
q.expect_many(EventPattern('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
iq_type='get'),
EventPattern('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
iq_type='set'),
EventPattern('stream-presence'))
for show, statuses in test_statuses.items():
shared_show, _ = _show_to_shared_status_show(show)
expected_list = stream.shared_status_lists[shared_show]
for status in statuses:
_test_local_status(q, conn, stream, status, show)
expected_list = [status] + expected_list[:max_statuses - 1]
assertEquals(expected_list, stream.shared_status_lists[shared_show])
开发者ID:Thaodan,项目名称:telepathy-gabble,代码行数:27,代码来源:shared-status.py
示例6: test
def test(q, bus, conn, stream):
# Request a sidecar thate we support before we're connected; it should just
# wait around until we're connected.
call_async(q, conn.Future, 'EnsureSidecar', PLUGIN_IFACE)
conn.Connect()
q.expect('dbus-signal', signal='StatusChanged',
args=[cs.CONN_STATUS_CONNECTED, cs.CSR_REQUESTED])
# Now we're connected, the call we made earlier should return.
path, props = q.expect('dbus-return', method='EnsureSidecar').value
# This sidecar doesn't even implement get_immutable_properties; it
# should just get the empty dict filled in for it.
assertEquals({}, props)
gateways_iface = dbus.Interface(bus.get_object(conn.bus_name, path),
PLUGIN_IFACE)
test_success(q, gateways_iface, stream)
test_conflict(q, gateways_iface, stream)
test_not_acceptable(q, gateways_iface, stream)
call_async(q, conn, 'Disconnect')
q.expect_many(
EventPattern('dbus-signal', signal='StatusChanged',
args=[cs.CONN_STATUS_DISCONNECTED, cs.CSR_REQUESTED]),
EventPattern('stream-closed'),
)
stream.sendFooter()
q.expect('dbus-return', method='Disconnect')
开发者ID:jku,项目名称:telepathy-gabble,代码行数:32,代码来源:gateways.py
示例7: test_invisible_on_connect
def test_invisible_on_connect(q, bus, conn, stream):
props = conn.Properties.GetAll(cs.CONN_IFACE_SIMPLE_PRESENCE)
assertNotEquals({}, props['Statuses'])
presence_event_pattern = EventPattern('stream-presence')
q.forbid_events([presence_event_pattern])
conn.SimplePresence.SetPresence("hidden", "")
conn.Connect()
stream.handle_get_all_privacy_lists(q, bus, conn, ['invisible'])
get_list = q.expect('stream-iq', query_ns=ns.PRIVACY, iq_type='get')
list_node = xpath.queryForNodes('//list', get_list.query)[0]
assertEquals('invisible', list_node['name'])
stream.send_privacy_list(get_list.stanza,
[elem('item', action='deny', order='1')(elem('presence-out'))])
set_active = q.expect('stream-iq', query_ns=ns.PRIVACY, iq_type='set')
active = xpath.queryForNodes('//active', set_active.query)[0]
assertEquals('invisible', active['name'])
acknowledge_iq(stream, set_active.stanza)
q.unforbid_events([presence_event_pattern])
q.expect('dbus-signal', signal='StatusChanged',
args=[cs.CONN_STATUS_CONNECTED, cs.CSR_REQUESTED])
开发者ID:jku,项目名称:telepathy-gabble,代码行数:30,代码来源:invisible_xep_0126.py
示例8: run_cancel_test
def run_cancel_test(q, bus, conn, stream):
jp = JingleProtocol031 ()
jt = JingleTest2(jp, conn, q, stream, '[email protected]', muc + '/bob')
jt.prepare()
for x in xrange (0, 10):
(path, props) = create_muji_channel (q, conn, stream, muc, x > 0)
channel = bus.get_object (conn.bus_name, path)
contents = channel.Get (cs.CHANNEL_TYPE_CALL, "Contents",
dbus_interface = dbus.PROPERTIES_IFACE)
content = bus.get_object (conn.bus_name, contents[0])
md = jt.get_call_audio_md_dbus()
check_and_accept_offer (q, bus, conn, content, md)
# Accept the channel
channel.Accept()
e = q.expect('stream-presence', to = muc + "/test")
mujinode = xpath.queryForNodes("/presence/muji/preparing", e.stanza)
assertNotEquals(None, mujinode)
channel.Hangup(0, "", "",
dbus_interface=cs.CHANNEL_TYPE_CALL)
e = q.expect('stream-presence', to = muc + "/test")
mujinode = xpath.queryForNodes("/presence/muji/preparing", e.stanza)
assertEquals(None, mujinode)
if x % 2 == 0:
channel.Close()
开发者ID:mlundblad,项目名称:telepathy-gabble,代码行数:33,代码来源:call-muc-re-re-request.py
示例9: run_test
def run_test(q, bus, conn, stream, jt, decloak_allowed):
"""
Requests streams on a media channel to jt.remote_jid without having their
presence at all.
"""
call_async(q, conn.Requests, 'CreateChannel',
{ cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_CALL,
cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
cs.TARGET_ID: jt.peer,
cs.CALL_INITIAL_AUDIO: True,
cs.CALL_INITIAL_VIDEO: False,
})
e = q.expect('stream-presence',
to=jt.peer_bare_jid, presence_type=None)
nodes = [ node for node in e.stanza.elements(uri=ns.TEMPPRES, name='temppres') ]
assertLength(1, nodes)
assertEquals('media', nodes[0].getAttribute('reason'))
if decloak_allowed:
jt.send_presence_and_caps()
# RequestStreams should now happily complete
q.expect('dbus-return', method='CreateChannel')
else:
q.expect('dbus-error', method='CreateChannel',
name=cs.OFFLINE)
开发者ID:Thaodan,项目名称:telepathy-gabble,代码行数:28,代码来源:decloak-peer.py
示例10: _cb_disco_iq
def _cb_disco_iq(self, iq):
nodes = xpath.queryForNodes("/iq/query", iq)
query = nodes[0]
if query.getAttribute('node') is None:
return
node = query.attributes['node']
ver = node.replace("http://telepathy.freedesktop.org/caps#", "")
if iq.getAttribute('type') == 'result':
if FileTransferTest.caps_identities is None or \
FileTransferTest.caps_features is None:
identity_nodes = xpath.queryForNodes('/iq/query/identity', iq)
assertLength(1, identity_nodes)
identity_node = identity_nodes[0]
identity_category = identity_node['category']
identity_type = identity_node['type']
identity_name = identity_node['name']
identity = '%s/%s//%s' % (identity_category, identity_type,
identity_name)
FileTransferTest.caps_identities = [identity]
FileTransferTest.caps_features = []
for feature in xpath.queryForNodes('/iq/query/feature', iq):
FileTransferTest.caps_features.append(feature['var'])
# Check if the hash matches the announced capabilities
assertEquals(compute_caps_hash(FileTransferTest.caps_identities,
FileTransferTest.caps_features,
{}), ver)
if ver == FileTransferTest.caps_ft:
caps_share = compute_caps_hash(FileTransferTest.caps_identities,
FileTransferTest.caps_features + \
[ns.GOOGLE_FEAT_SHARE],
{})
n = query.attributes['node'].replace(ver, caps_share)
query.attributes['node'] = n
for feature in xpath.queryForNodes('/iq/query/feature', iq):
query.children.remove(feature)
for f in FileTransferTest.caps_features + [ns.GOOGLE_FEAT_SHARE]:
el = domish.Element((None, 'feature'))
el['var'] = f
query.addChild(el)
elif iq.getAttribute('type') == 'get':
caps_share = compute_caps_hash(FileTransferTest.caps_identities,
FileTransferTest.caps_features + \
[ns.GOOGLE_FEAT_SHARE],
{})
if ver == caps_share:
n = query.attributes['node'].replace(ver,
FileTransferTest.caps_ft)
query.attributes['node'] = n
开发者ID:jku,项目名称:telepathy-gabble,代码行数:60,代码来源:file_transfer_helper.py
示例11: test
def test(q, bus, conn, stream):
conn.Connect()
q.expect('dbus-signal', signal='StatusChanged', args=[0, 1])
CHANNEL_NAME = '#idletest'
call_async(q, conn.Requests, 'CreateChannel',
{ CHANNEL_TYPE: CHANNEL_TYPE_TEXT,
TARGET_HANDLE_TYPE: HT_ROOM,
TARGET_ID: CHANNEL_NAME })
ret = q.expect('dbus-return', method='CreateChannel')
q.expect('dbus-signal', signal='MembersChanged')
chan = bus.get_object(conn.bus_name, ret.value[0])
text_chan = dbus.Interface(chan, CHANNEL_TYPE_TEXT)
# send a whole bunch of messages in a row and make sure they get delivered
# in the proper order
NUM_MESSAGES = 4
for i in range(NUM_MESSAGES):
call_async(q, text_chan, 'Send', 0, str(i))
for i in range(NUM_MESSAGES):
message = q.expect('stream-PRIVMSG')
assertEquals([CHANNEL_NAME, str(i)], message.data)
call_async(q, conn, 'Disconnect')
开发者ID:freedesktop-unofficial-mirror,项目名称:telepathy__telepathy-idle,代码行数:26,代码来源:message-order.py
示例12: accept
def accept(self):
if self.incoming:
self.chan.Call1.SetQueued()
o = self.q.expect_many(
EventPattern('sip-response', call_id=self.context.call_id,
code=182),
EventPattern('dbus-signal', signal='CallStateChanged'))
assertEquals(cs.CALL_STATE_INITIALISED, o[1].args[0])
assertEquals(cs.CALL_FLAG_LOCALLY_QUEUED, o[1].args[1])
self.chan.Call1.SetRinging()
o = self.q.expect_many(
EventPattern('sip-response', call_id=self.context.call_id,
code=180),
EventPattern('dbus-signal', signal='CallStateChanged'))
assertEquals(cs.CALL_STATE_INITIALISED, o[1].args[0])
assertEquals(cs.CALL_FLAG_LOCALLY_RINGING, o[1].args[1])
else:
# Send Ringing
self.context.pr_respond(self.invite_event, 180)
o = self.q.expect('dbus-signal', signal='CallMembersChanged')
assertEquals(cs.CALL_MEMBER_FLAG_RINGING,
o.args[0][self.remote_handle])
return calltest.CallTest.accept(self)
开发者ID:freedesktop-unofficial-mirror,项目名称:telepathy__telepathy-rakia,代码行数:28,代码来源:ringing-queued.py
示例13: test_error
def test_error(q, bus, conn, stream):
assertContains(cs.CONN_IFACE_POWER_SAVING,
conn.Get(cs.CONN, "Interfaces",
dbus_interface=cs.PROPERTIES_IFACE))
assertEquals (False, conn.Get(cs.CONN_IFACE_POWER_SAVING,
"PowerSavingActive",
dbus_interface=cs.PROPERTIES_IFACE))
call_async(q, conn.PowerSaving, 'SetPowerSaving', True)
stanza = expect_command(q, 'enable')
error = domish.Element((None, 'error'))
error.addElement((ns.STANZA, 'service-unavailable'))
send_error_reply(stream, stanza, error)
q.expect('dbus-error', method='SetPowerSaving', name=cs.NOT_AVAILABLE)
# Power saving state should remain false
assertEquals (False, conn.Get(cs.CONN_IFACE_POWER_SAVING,
"PowerSavingActive",
dbus_interface=cs.PROPERTIES_IFACE))
开发者ID:jku,项目名称:telepathy-gabble,代码行数:25,代码来源:power-save.py
示例14: tube_then_text
def tube_then_text(q, bus, conn):
jid = 'test-muc'
connect(q, bus, conn)
# first let's get a stream tube
tube_chan, _, _ = stream_tube(q, bus, conn, 'CreateChannel', jid)
# now let's try and ensure the text channel which should happen
# immediately
request_text_channel(q, bus, conn, 'EnsureChannel', jid)
ret = q.expect('dbus-return', method='EnsureChannel')
yours, text_path, text_props = ret.value
assertEquals(True, yours)
assertEquals(cs.CHANNEL_TYPE_TEXT, text_props[cs.CHANNEL_TYPE])
new_sig = q.expect('dbus-signal', signal='NewChannels')
channels = new_sig.args[0]
assertEquals(1, len(channels))
path, props = channels[0]
assertEquals(text_path, path)
assertEquals(text_props, props)
开发者ID:freedesktop-unofficial-mirror,项目名称:telepathy__telepathy-salut,代码行数:26,代码来源:only-text-muc-when-needed.py
示例15: stop_sending
def stop_sending(self, content):
self.sending = False
content.stream.SetSending(False)
self.q.expect('dbus-signal', signal='SendingStateChanged',
args=[cs.CALL_STREAM_FLOW_STATE_PENDING_STOP],
path=content.stream.__dbus_object_path__)
content.stream.Media.CompleteSendingStateChange(
cs.CALL_STREAM_FLOW_STATE_STOPPED)
o = self.q.expect_many(
EventPattern('dbus-signal', signal='SendingStateChanged',
args=[cs.CALL_STREAM_FLOW_STATE_STOPPED],
path=content.stream.__dbus_object_path__),
EventPattern('dbus-signal', signal='LocalSendingStateChanged',
path=content.stream.__dbus_object_path__),
EventPattern('sip-invite'))
assertEquals(cs.CALL_SENDING_STATE_NONE, o[1].args[0])
assertEquals(self.self_handle, o[1].args[1][0])
reinvite_event = o[2]
assertContains('a=recvonly', reinvite_event.sip_message.body)
self.context.check_call_sdp(reinvite_event.sip_message.body)
body = reinvite_event.sip_message.body.replace(
'recvonly', self.receiving and 'sendonly' or 'inactive')
self.context.accept(reinvite_event.sip_message, body)
ack_cseq = "%s ACK" % reinvite_event.cseq.split()[0]
self.q.expect('sip-ack', cseq=ack_cseq)
开发者ID:freedesktop-unofficial-mirror,项目名称:telepathy__telepathy-rakia,代码行数:33,代码来源:direction-change.py
示例16: test_content_addition
def test_content_addition(self):
path = self.chan.AddContent("Webcam", cs.CALL_MEDIA_TYPE_VIDEO,
cs.MEDIA_STREAM_DIRECTION_BIDIRECTIONAL,
dbus_interface=cs.CHANNEL_TYPE_CALL)
content = self.bus.get_object(self.conn.bus_name, path)
content_properties = content.GetAll(cs.CALL_CONTENT,
dbus_interface=dbus.PROPERTIES_IFACE)
assertEquals(cs.CALL_DISPOSITION_NONE,
content_properties["Disposition"])
#assertEquals(self_handle, content_properties["Creator"])
assertContains("Webcam", content_properties["Name"])
md = self.jt2.get_call_video_md_dbus()
self.check_and_accept_offer(content, md)
cstream = self.bus.get_object(self.conn.bus_name,
content_properties["Streams"][0])
candidates = self.jt2.get_call_remote_transports_dbus()
cstream.AddCandidates(candidates,
dbus_interface=cs.CALL_STREAM_IFACE_MEDIA)
self.q.expect('stream-iq',
predicate=self.jp.action_predicate('content-add'))
content.Remove(dbus_interface=cs.CALL_CONTENT)
self.q.expect('stream-iq',
predicate=self.jp.action_predicate('content-remove'))
开发者ID:Thaodan,项目名称:telepathy-gabble,代码行数:28,代码来源:call-basics.py
示例17: sending_failed
def sending_failed(self, content):
self.sending = False
content.stream.Media.ReportSendingFailure(
cs.CALL_STATE_CHANGE_REASON_MEDIA_ERROR, "", "sending error")
o = self.q.expect_many(
EventPattern('dbus-signal', signal='SendingStateChanged',
args=[cs.CALL_STREAM_FLOW_STATE_STOPPED],
path=content.stream.__dbus_object_path__),
EventPattern('dbus-signal', signal='LocalSendingStateChanged',
path=content.stream.__dbus_object_path__,
predicate=lambda e: e.args[0] == 0),
EventPattern('sip-invite'))
assertEquals(cs.CALL_SENDING_STATE_NONE, o[1].args[0])
assertEquals(self.self_handle, o[1].args[1][0])
reinvite_event = o[2]
assertContains('a=recvonly', reinvite_event.sip_message.body)
self.context.check_call_sdp(reinvite_event.sip_message.body)
body = reinvite_event.sip_message.body.replace(
'recvonly', self.receiving and 'sendonly' or 'inactive')
self.context.accept(reinvite_event.sip_message, body)
ack_cseq = "%s ACK" % reinvite_event.cseq.split()[0]
self.q.expect('sip-ack', cseq=ack_cseq)
self.start_sending(content)
开发者ID:freedesktop-unofficial-mirror,项目名称:telepathy__telepathy-rakia,代码行数:31,代码来源:direction-change.py
示例18: test_new_channel
def test_new_channel(q, bus, conn, target_uri, initiator_uri, requested):
event = q.expect('dbus-signal', signal='NewChannels')
path, props = event.args[0][0]
assertEquals(cs.CHANNEL_TYPE_TEXT, props[cs.CHANNEL_TYPE])
assertEquals(cs.HT_CONTACT, props[cs.TARGET_HANDLE_TYPE])
handle = props[cs.TARGET_HANDLE]
obj = bus.get_object(conn._named_service, path)
initiator_handle = conn.get_contact_handle_sync(initiator_uri)
text_props = obj.GetAll(cs.CHANNEL,
dbus_interface='org.freedesktop.DBus.Properties')
assert text_props['ChannelType'] == cs.CHANNEL_TYPE_TEXT, text_props
assert 'Interfaces' in text_props, text_props
assertSameSets((cs.CHANNEL_IFACE_MESSAGES, cs.CHANNEL_IFACE_DESTROYABLE),
text_props['Interfaces'])
assert 'TargetHandle' in text_props, text_props
assert text_props['TargetHandle'] == handle, \
(text_props, handle)
assert 'TargetHandleType' in text_props, text_props
assert text_props['TargetHandleType'] == 1, text_props
assert text_props['TargetID'] == target_uri, text_props
assert text_props['InitiatorHandle'] == initiator_handle, \
(text_props, initiator_handle)
assert text_props['InitiatorID'] == initiator_uri, \
(text_props, initiator_uri)
assert 'Requested' in text_props, text_props
assert text_props['Requested'] == requested, text_props
return obj, handle
开发者ID:freedesktop-unofficial-mirror,项目名称:telepathy__telepathy-rakia,代码行数:30,代码来源:test-message.py
示例19: test_invisible_on_connect_fail
def test_invisible_on_connect_fail(q, bus, conn, stream):
props = conn.Properties.GetAll(cs.CONN_IFACE_SIMPLE_PRESENCE)
assertNotEquals({}, props['Statuses'])
presence_event_pattern = EventPattern('stream-presence')
q.forbid_events([presence_event_pattern])
conn.SimplePresence.SetPresence("hidden", "")
conn.Connect()
stream.handle_get_all_privacy_lists(q, bus, conn)
create_list = q.expect('stream-iq', query_ns=ns.PRIVACY, iq_type='set')
# Check its name
assertNotEquals([],
xpath.queryForNodes('/query/list/item/presence-out', create_list.query))
acknowledge_iq(stream, create_list.stanza)
set_active = q.expect('stream-iq', query_ns=ns.PRIVACY, iq_type='set')
active = xpath.queryForNodes('//active', set_active.query)[0]
assertEquals('invisible', active['name'])
send_error_reply(stream, set_active.stanza)
q.unforbid_events([presence_event_pattern])
# Darn! At least we should have our presence set to DND.
q.expect_many(
EventPattern('dbus-signal', signal='PresencesChanged',
interface=cs.CONN_IFACE_SIMPLE_PRESENCE,
args=[{1: (6, 'dnd', '')}]),
EventPattern('dbus-signal', signal='StatusChanged',
args=[cs.CONN_STATUS_CONNECTED, cs.CSR_REQUESTED]))
开发者ID:jku,项目名称:telepathy-gabble,代码行数:34,代码来源:invisible_xep_0126.py
示例20: check_and_accept_offer
def check_and_accept_offer (q, bus, conn, content, md, in_remote_handle = 0,
offer_path = None, md_changed = True):
[path, remote_md] = content.Get(cs.CALL_CONTENT_IFACE_MEDIA,
"MediaDescriptionOffer", dbus_interface=dbus.PROPERTIES_IFACE)
remote_handle = remote_md[cs.CALL_CONTENT_MEDIADESCRIPTION + '.RemoteContact']
if in_remote_handle != 0:
assertEquals(remote_handle, in_remote_handle)
if offer_path != None:
assertEquals (offer_path, path)
assertNotEquals ("/", path)
offer = bus.get_object (conn.bus_name, path)
codecmap_property = offer.Get (cs.CALL_CONTENT_MEDIADESCRIPTION,
"Codecs", dbus_interface=dbus.PROPERTIES_IFACE)
assertEquals (remote_md[cs.CALL_CONTENT_MEDIADESCRIPTION + '.Codecs'], codecmap_property)
offer.Accept (md, dbus_interface=cs.CALL_CONTENT_MEDIADESCRIPTION)
current_md = content.Get(cs.CALL_CONTENT_IFACE_MEDIA,
"LocalMediaDescriptions", dbus_interface=dbus.PROPERTIES_IFACE)
assertEquals (md, current_md[remote_handle])
if md_changed:
o = q.expect ('dbus-signal', signal='LocalMediaDescriptionChanged')
assertEquals ([md], o.args)
开发者ID:mlundblad,项目名称:telepathy-gabble,代码行数:31,代码来源:callutils.py
注:本文中的servicetest.assertEquals函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论