本文整理汇总了Python中vumi.transports.mtn_nigeria.tests.utils.mk_packet函数的典型用法代码示例。如果您正苦于以下问题：Python mk_packet函数的具体用法？Python mk_packet怎么用？Python mk_packet使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了mk_packet函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_periodic_client_enquire_link
def test_periodic_client_enquire_link(self):
request_body_a = (
"<ENQRequest>"
"<requestId>0</requestId>"
"<enqCmd>ENQUIRELINK</enqCmd>"
"</ENQRequest>")
expected_request_packet_a = utils.mk_packet('0', request_body_a)
response_body_a = (
"<ENQResponse>"
"<requestId>0</requestId>"
"<enqCmd>ENQUIRELINKRSP</enqCmd>"
"</ENQResponse>")
response_packet_a = utils.mk_packet('0', response_body_a)
self.server.responses[expected_request_packet_a] = response_packet_a
request_body_b = (
"<ENQRequest>"
"<requestId>1</requestId>"
"<enqCmd>ENQUIRELINK</enqCmd>"
"</ENQRequest>")
expected_request_packet_b = utils.mk_packet('1', request_body_b)
response_body_b = (
"<ENQResponse>"
"<requestId>1</requestId>"
"<enqCmd>ENQUIRELINKRSP</enqCmd>"
"</ENQResponse>")
response_packet_b = utils.mk_packet('1', response_body_b)
self.server.responses[expected_request_packet_b] = response_packet_b
clock = Clock()
t0 = clock.seconds()
self.client.clock = clock
self.client.enquire_link_interval = 120
self.client.timeout_period = 20
self.client.authenticated = True
self.client.start_periodic_enquire_link()
# advance to just after the first enquire link request
clock.advance(0.01)
self.assert_next_timeout(t0 + 20)
# wait for the first enquire link response
yield self.client.wait_for_data()
self.assert_timeout_cancelled()
# advance to just after the second enquire link request
clock.advance(120.01)
self.assert_next_timeout(t0 + 140)
# wait for the second enquire link response
yield self.client.wait_for_data()
self.assert_timeout_cancelled()
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:54，代码来源:test_xml_over_tcp.py
示例2: test_partial_data_received
def test_partial_data_received(self):
body_a = "<DummyPacket><someParam>123</someParam></DummyPacket>"
body_b = "<DummyPacket><someParam>456</someParam></DummyPacket>"
# add a full first packet, then concatenate a sliced version of a
# second packet
data = utils.mk_packet('0', body_a)
data += utils.mk_packet('1', body_b)[:12]
self.client.authenticated = True
self.server.send_data(data)
yield self.client.wait_for_data()
self.assertEqual(
self.client.received_dummy_packets,
[('0', {'someParam': '123'})])
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:15，代码来源:test_xml_over_tcp.py
示例3: test_contiguous_packets_received
def test_contiguous_packets_received(self):
body_a = "<DummyPacket><someParam>123</someParam></DummyPacket>"
body_b = "<DummyPacket><someParam>456</someParam></DummyPacket>"
data = utils.mk_packet('0', body_a)
data += utils.mk_packet('1', body_b)
self.client.authenticated = True
self.server.send_data(data)
yield self.client.wait_for_data()
self.assertEqual(
self.client.received_dummy_packets, [
('0', {'someParam': '123'}),
('1', {'someParam': '456'}),
])
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:15，代码来源:test_xml_over_tcp.py
示例4: test_ending_session_data_response
def test_ending_session_data_response(self):
body = (
"<USSDResponse>"
"<requestId>1291850641</requestId>"
"<msisdn>27845335367</msisdn>"
"<starCode>123</starCode>"
"<clientId>123</clientId>"
"<phase>2</phase>"
"<msgtype>6</msgtype>"
"<dcs>15</dcs>"
"<userdata>*123#</userdata>"
"<EndofSession>1</EndofSession>"
"<delvrpt>0</delvrpt>"
"</USSDResponse>"
)
expected_packet = utils.mk_packet('0', body)
self.client.authenticated = True
self.client.send_data_response(
session_id='0',
request_id='1291850641',
star_code='123',
client_id='123',
msisdn='27845335367',
user_data='*123#',
end_session=True)
received_packet = yield self.server.wait_for_data()
self.assertEqual(expected_packet, received_packet)
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:29，代码来源:test_xml_over_tcp.py
示例5: test_timeout
def test_timeout(self):
request_body = (
"<ENQRequest>"
"<requestId>0</requestId>"
"<enqCmd>ENQUIRELINK</enqCmd>"
"</ENQRequest>")
expected_request_packet = utils.mk_packet('0', request_body)
clock = Clock()
self.client.clock = clock
self.client.enquire_link_interval = 120
self.client.timeout_period = 20
self.client.authenticated = True
self.client.start_periodic_enquire_link()
# wait for the first enquire link request
received_request_packet = yield self.server.wait_for_data()
self.assertEqual(expected_request_packet, received_request_packet)
# advance to just before the timeout should occur
clock.advance(19.9)
self.assertFalse(self.client.disconnected)
# advance to just after the timeout should occur
clock.advance(0.1)
self.assertTrue(self.client.disconnected)
self.assert_in_log(
'msg',
"No enquire link response received after 20 seconds, "
"disconnecting")
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:30，代码来源:test_xml_over_tcp.py
示例6: mk_error_response
def mk_error_response(self, session_id, request_id, error_code):
body = (
"<USSDError>"
"<requestId>%s</requestId>"
"<errorCode>%s</errorCode>"
"</USSDError>" % (request_id, error_code))
return utils.mk_packet(session_id, body)
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:7，代码来源:test_mtn_nigeria.py
示例7: test_data_request_handling
def test_data_request_handling(self):
body = (
"<USSDRequest>"
"<requestId>1291850641</requestId>"
"<msisdn>27845335367</msisdn>"
"<starCode>123</starCode>"
"<clientId>123</clientId>"
"<phase>2</phase>"
"<dcs>15</dcs>"
"<userdata>*123#</userdata>"
"<msgtype>1</msgtype>"
"<EndofSession>0</EndofSession>"
"</USSDRequest>"
)
packet = utils.mk_packet('0', body)
self.client.authenticated = True
self.server.send_data(packet)
yield self.client.wait_for_data()
expected_params = {
'requestId': '1291850641',
'msisdn': '27845335367',
'starCode': '123',
'clientId': '123',
'phase': '2',
'dcs': '15',
'userdata': '*123#',
'msgtype': '1',
'EndofSession': '0',
}
self.assertEqual(
self.client.received_data_request_packets,
[('0', expected_params)])
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:33，代码来源:test_xml_over_tcp.py
示例8: test_error_response_validation
def test_error_response_validation(self):
bad_packet = utils.mk_packet(
'0', "<USSDError><requestId>0</requestId></USSDError>")
self.server.send_data(bad_packet)
yield self.client.wait_for_data()
self.assert_in_log(
'err',
"(208) Invalid Message: Missing mandatory fields in received "
"packet: %s" % ['errorCode'])
received_packet = yield self.server.wait_for_data()
self.assertEqual(received_packet, utils.mk_packet(
'0',
"<USSDError>"
"<requestId>0</requestId>"
"<errorCode>208</errorCode>"
"</USSDError>"))
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:18，代码来源:test_xml_over_tcp.py
示例9: test_unknown_packet_handling
def test_unknown_packet_handling(self):
request_body = (
"<UnknownPacket><requestId>0</requestId></UnknownPacket>")
request_packet = utils.mk_packet('0', request_body)
response_body = (
"<USSDError>"
"<requestId>0</requestId>"
"<errorCode>208</errorCode>"
"</USSDError>")
expected_response_packet = utils.mk_packet('0', response_body)
self.server.send_data(request_packet)
yield self.client.wait_for_data()
response_packet = yield self.server.wait_for_data()
self.assertEqual(expected_response_packet, response_packet)
self.assert_in_log(
'err', "Packet of an unknown type received: UnknownPacket")
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:19，代码来源:test_xml_over_tcp.py
示例10: test_packet_parsing_for_packets_with_wierd_bodies
def test_packet_parsing_for_packets_with_wierd_bodies(self):
data = utils.mk_packet('0', "<BadPacket>")
self.client.authenticated = True
self.server.send_data(data)
yield self.client.wait_for_data()
err_msg = self.logs['err'][0]
self.assertTrue("Error parsing packet" in err_msg)
self.assertTrue(('%r' % (data,)) in err_msg)
self.assertTrue(self.client.disconnected)
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:10，代码来源:test_xml_over_tcp.py
示例11: test_enquire_link_request_validation
def test_enquire_link_request_validation(self):
bad_packet = utils.mk_packet(
'0', "<ENQRequest><requestId>0</requestId></ENQRequest>")
self.client.authenticated = True
self.server.send_data(bad_packet)
yield self.client.wait_for_data()
self.assert_in_log(
'err',
"(208) Invalid Message: Missing mandatory fields in received "
"packet: %s" % ['enqCmd'])
received_packet = yield self.server.wait_for_data()
self.assertEqual(received_packet, utils.mk_packet(
'0',
"<USSDError>"
"<requestId>0</requestId>"
"<errorCode>208</errorCode>"
"</USSDError>"))
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:19，代码来源:test_xml_over_tcp.py
示例12: test_server_enquire_link
def test_server_enquire_link(self):
request_body = (
"<ENQRequest>"
"<requestId>0</requestId>"
"<enqCmd>ENQUIRELINK</enqCmd>"
"</ENQRequest>")
request_packet = utils.mk_packet('0', request_body)
response_body = (
"<ENQResponse>"
"<requestId>0</requestId>"
"<enqCmd>ENQUIRELINKRSP</enqCmd>"
"</ENQResponse>")
expected_response_packet = utils.mk_packet('0', response_body)
self.client.authenticated = True
self.server.send_data(request_packet)
response_packet = yield self.server.wait_for_data()
self.assertEqual(expected_response_packet, response_packet)
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:19，代码来源:test_xml_over_tcp.py
示例13: test_authentication
def test_authentication(self):
request_body = (
"<AUTHRequest>"
"<requestId>0</requestId>"
"<userName>root</userName>"
"<passWord>toor</passWord>"
"<applicationId>1029384756</applicationId>"
"</AUTHRequest>")
expected_request_packet = utils.mk_packet('0', request_body)
response_body = (
"<AUTHResponse>"
"<requestId>0</requestId>"
"<authMsg>SUCCESS</authMsg>"
"</AUTHResponse>")
response_packet = utils.mk_packet('a', response_body)
self.server.responses[expected_request_packet] = response_packet
self.client.login()
yield self.client.wait_for_data()
self.assertTrue(self.client.authenticated)
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:21，代码来源:test_xml_over_tcp.py
示例14: test_data_request_validation
def test_data_request_validation(self):
bad_packet = utils.mk_packet(
'0', "<USSDRequest><requestId>0</requestId></USSDRequest>")
self.client.authenticated = True
self.server.send_data(bad_packet)
yield self.client.wait_for_data()
missing_fields = ['userdata', 'msisdn', 'clientId', 'starCode',
'msgtype', 'phase', 'dcs']
self.assert_in_log(
'err',
"(208) Invalid Message: Missing mandatory fields in received "
"packet: %s" % sorted(missing_fields))
received_packet = yield self.server.wait_for_data()
self.assertEqual(received_packet, utils.mk_packet(
'0',
"<USSDError>"
"<requestId>0</requestId>"
"<errorCode>208</errorCode>"
"</USSDError>"))
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:22，代码来源:test_xml_over_tcp.py
示例15: test_packet_received_before_auth
def test_packet_received_before_auth(self):
request_body = (
"<DummyPacket>"
"<requestId>0</requestId>"
"</DummyPacket>")
request_packet = utils.mk_packet('0', request_body)
response_body = (
"<USSDError>"
"<requestId>0</requestId>"
"<errorCode>207</errorCode>"
"</USSDError>")
expected_response_packet = utils.mk_packet('0', response_body)
self.server.send_data(request_packet)
yield self.client.wait_for_data()
response_packet = yield self.server.wait_for_data()
self.assertEqual(expected_response_packet, response_packet)
self.assert_in_log(
'err',
"'DummyPacket' packet received before client authentication was "
"completed")
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:23，代码来源:test_xml_over_tcp.py
示例16: test_authentication_error_handling
def test_authentication_error_handling(self):
request_body = (
"<AUTHRequest>"
"<requestId>0</requestId>"
"<userName>root</userName>"
"<passWord>toor</passWord>"
"<applicationId>1029384756</applicationId>"
"</AUTHRequest>")
expected_request_packet = utils.mk_packet('0', request_body)
response_body = (
"<AUTHError>"
"<requestId>0</requestId>"
"<authMsg>FAILURE</authMsg>"
"<errorCode>001</errorCode>"
"</AUTHError>")
response_packet = utils.mk_packet('0', response_body)
self.server.responses[expected_request_packet] = response_packet
self.client.login()
yield self.client.wait_for_data()
self.assertFalse(self.client.authenticated)
self.assertTrue(self.client.disconnected)
self.assert_in_log('err', 'Login failed, disconnecting')
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:24，代码来源:test_xml_over_tcp.py
示例17: test_error_response_handling_for_unknown_codes
def test_error_response_handling_for_unknown_codes(self):
body = (
"<USSDError>"
"<requestId>0</requestId>"
"<errorCode>1337</errorCode>"
"<errorMsg>Some Reason</errorMsg>"
"</USSDError>"
)
error_packet = utils.mk_packet('0', body)
self.server.send_data(error_packet)
yield self.client.wait_for_data()
self.assert_in_log(
'err',
"Server sent error message: (1337) Unknown Code: "
"Some Reason")
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:16，代码来源:test_xml_over_tcp.py
示例18: test_packets_split_over_socket_reads
def test_packets_split_over_socket_reads(self):
body = "<DummyPacket><someParam>123</someParam></DummyPacket>"
data = utils.mk_packet('0', body)
split_position = int(len(data) / 2)
self.client.authenticated = True
self.server.send_data(data[:split_position])
yield self.client.wait_for_data()
self.server.send_data(data[split_position:])
yield self.client.wait_for_data()
self.assertEqual(
self.client.received_dummy_packets,
[('0', {'someParam': '123'})])
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:16，代码来源:test_xml_over_tcp.py
示例19: test_data_response_handling
def test_data_response_handling(self):
body = (
"<USSDResponse>"
"<requestId>1291850641</requestId>"
"<msisdn>27845335367</msisdn>"
"<starCode>123</starCode>"
"<clientId>123</clientId>"
"<phase>2</phase>"
"<dcs>15</dcs>"
"<userdata>Session closed due to cause 0</userdata>"
"<msgtype>1</msgtype>"
"<EndofSession>1</EndofSession>"
"</USSDResponse>"
)
packet = utils.mk_packet('0', body)
self.client.authenticated = True
self.server.send_data(packet)
yield self.client.wait_for_data()
self.assert_in_log(
'msg', "Received spurious USSDResponse message, ignoring.")
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:21，代码来源:test_xml_over_tcp.py
示例20: mk_data_request
def mk_data_request(self, session_id, **kw):
params = self.REQUEST_PARAMS.copy()
params.update(kw)
return utils.mk_packet(session_id, self.REQUEST_BODY % params)
开发者ID:AndrewCvekl，项目名称:vumi，代码行数:4，代码来源:test_mtn_nigeria.py
注：本文中的vumi.transports.mtn_nigeria.tests.utils.mk_packet函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论