本文整理汇总了Python中microgear.client.create函数的典型用法代码示例。如果您正苦于以下问题：Python create函数的具体用法？Python create怎么用？Python create使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了create函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: testCode7Case5
def testCode7Case5(self):
"""publish invalid topic - no slash"""
try:
print("run helper...")
code = str(61)
args = ['python', 'helper.py', code]
p = subprocess.Popen(args, cwd=(helper_dir))
time.sleep(connect_worst_timeout)
self.invalidTopic = "firstTopic"
client.create(self.gearkey, self.gearsecret, self.appid)
client.on_connect = MagicMock()
client.connect()
time.sleep(connect_timeout)
self.assertTrue(client.on_connect.called)
self.connected = True
self.assertEqual(client.on_connect.call_count, 1)
client.publish(self.invalidTopic, self.message)
time.sleep(message_timeout)
receiver_file = open(os.path.join(os.getcwd(),"receiver.txt"), "r")
received_message = receiver_file.read()
receiver_file.close()
if(received_message == self.message):
self.received = True
self.assertFalse(self.received)
self.assertTrue(client.on_connect.call_count > 1)
p.kill()
except Exception as e:
p.kill()
raise Exception(e.args)
开发者ID:kki32，项目名称:microgearlib_test_python，代码行数:35，代码来源:all.py
示例2: connectTo
def connectTo():
gearkey = "qnlgzsPUUxYeyQP"
gearsecret = "1euJPvxybllEPQZzq2u9wpRJXDbjM7"
appid = "testNo3"
client.create(gearkey, gearsecret, appid, {'debugmode': "True"})
def on_connected():
print("connect")
def on_closed():
print("close")
def on_rejected():
print("reject")
def on_error():
print("error")
def on_message():
print("message")
def on_present():
print("present")
def on_absent():
print("absent")
client.on_connect = on_connected
client.on_error = on_error
client.on_present = on_present
client.on_absent = on_absent
client.on_rejected = on_rejected
client.on_closed = on_closed
client.on_message = on_message
logs = LogCapture()
client.connect()
print(logs)
logs.check(('root', 'DEBUG', 'Check stored token.'))
开发者ID:kki32，项目名称:MicroGearTestingSuite，代码行数:31，代码来源:testScript.py
示例3: testCode5Case2
def testCode5Case2(self):
"""subscribe same topic twice"""
try:
print('Code5Case2')
print("run helper...")
code = str(51)
args = ['python', 'helper.py', code]
p = subprocess.Popen(args, cwd=(helper_dir))
time.sleep(connect_worst_timeout)
client.create(self.gearkey, self.gearsecret, self.appid)
client.subscribe(self.topic)
client.subscribe(self.topic)
client.on_connect = MagicMock()
client.on_message = MagicMock()
client.connect()
time.sleep(connect_timeout)
self.assertTrue(client.on_connect.called)
self.connected = True
self.assertEqual(client.on_connect.call_count, 1)
time.sleep(message_timeout)
self.assertTrue(client.on_message.called)
client.on_message.assert_called_with(self.expectedTopic, self.expectedMessage)
p.kill()
#if fails due to assertion error
except Exception as e:
p.kill()
raise Exception(e.args)
开发者ID:kki32，项目名称:microgearlib_test_python，代码行数:30，代码来源:all.py
示例4: testCode8Case2
def testCode8Case2(self):
"""resettoken when have microgear.cache while microgear is offline"""
self.assertTrue(os.path.isfile(os.path.join(os.getcwd()+"/microgear.cache")))
self.assertIsNone(microgear.gearkey)
self.assertIsNone(microgear.gearsecret)
self.assertIsNone(microgear.appid)
client.create(self.gearkey, self.gearsecret, self.appid)
client.on_connect = MagicMock()
client.resettoken()
self.assertFalse(os.path.isfile(os.path.join(os.getcwd()+"/microgear.cache")))
client.connect()
time.sleep(connect_timeout)
self.assertTrue(client.on_connect.called)
self.assertEqual(client.on_connect.call_count, 1)
self.assertTrue(os.path.isfile(os.path.join(os.getcwd()+"/microgear.cache")))
# def main():
# pass
# #suite = unittest.TestSuite()
# #suite.addTest(TestChat("testCode4Case2"))
# #runner = unittest.TextTestRunner()
# #runner.run(suite)
# print(os.path.join(os.getcwd(),"microgear.cache"))
# unittest.main()
# if __name__ == '__main__':
# main()
开发者ID:kki32，项目名称:microgearlib_test_python，代码行数:33，代码来源:unitTestPython.py
示例5: testCode5Case6x2
def testCode5Case6x2(self):
"""subscribe invalid topic - no slash"""
try:
print('Code5Case6x2')
print("run helper...")
code = str(53)
args = ['python', 'helper.py', code]
p = subprocess.Popen(args, cwd=(helper_dir))
time.sleep(connect_worst_timeout)
self.topic = "firstTopic"
client.create(self.gearkey, self.gearsecret, self.appid)
client.subscribe(self.topic)
client.on_connect = MagicMock()
client.on_message = MagicMock()
client.connect()
time.sleep(connect_timeout)
self.assertTrue(client.on_connect.called)
self.connected = True
self.assertEqual(client.on_connect.call_count, 1)
time.sleep(message_timeout)
self.assertFalse(client.on_message.called)
p.kill()
#if fails due to assertion error
except Exception as e:
p.kill()
raise Exception(e.args)
开发者ID:kki32，项目名称:microgearlib_test_python，代码行数:31，代码来源:all.py
示例6: testCreateNetPieLabel
def testCreateNetPieLabel():
gearkey, gearsecret, appid = testCreateNetPie1();
client.create(gearkey, gearsecret, appid, {'label' : "Microgear Python"})
client.setname('logg')
client.connect()
print("Sleep for 90 seconds")
time.sleep(90)
开发者ID:kki32，项目名称:MicroGearTestingSuite，代码行数:7，代码来源:testCreate.py
示例7: testCreateNetPieScopeName
def testCreateNetPieScopeName():
gearkey, gearsecret, appid = testCreateNetPie1();
client.create(gearkey, gearsecret, appid, {'debugmode' : True, 'scope' : "name:logger"})
client.setname('logg')
client.connect()
print("Sleep for 90 seconds")
time.sleep(90)
开发者ID:kki32，项目名称:MicroGearTestingSuite，代码行数:7，代码来源:testCreate.py
示例8: testCode6Case1
def testCode6Case1(self):
"""unsubscribe the subscribed topic"""
print('Code6Case1')
try:
print("run helper...")
code = str(51)
args = ['python', 'helper.py', code]
p = subprocess.Popen(args, cwd=(helper_dir))
time.sleep(connect_worst_timeout)
client.create(self.gearkey, self.gearsecret, self.appid)
client.on_connect = MagicMock()
client.on_message = MagicMock()
client.connect()
time.sleep(connect_timeout)
self.assertTrue(client.on_connect.called)
self.connected = True
self.assertEqual(client.on_connect.call_count, 1)
client.subscribe(self.topic)
time.sleep(message_timeout)
self.assertTrue(client.on_message.called)
client.unsubscribe(self.topic)
client.on_message.reset_mock()
self.assertFalse(client.on_message.called)
time.sleep(message_timeout)
self.assertFalse(client.on_message.called)
p.kill()
#if fails due to assertion error
except Exception as e:
p.kill()
raise Exception(e.args)
开发者ID:kki32，项目名称:microgearlib_test_python，代码行数:35，代码来源:all.py
示例9: testCode6Case2
def testCode6Case2(self):
"""unsubscribe the topic before subscribe"""
print('Code6Case2')
print(microgear.gearkey)
try:
print("run helper...")
code = str(51)
args = ['python', 'helper.py', code]
p = subprocess.Popen(args, cwd=(helper_dir))
time.sleep(connect_worst_timeout)
print(microgear.gearkey)
client.create(self.gearkey, self.gearsecret, self.appid)
client.on_connect = MagicMock()
client.on_message = MagicMock()
client.connect()
time.sleep(connect_timeout)
self.assertTrue(client.on_connect.called)
self.connected = True
self.assertEqual(client.on_connect.call_count, 1)
self.assertFalse(client.on_message.called)
client.unsubscribe(self.topic)
self.assertFalse(client.on_message.called)
client.subscribe(self.topic)
time.sleep(message_timeout)
self.assertTrue(client.on_message.called)
client.on_message.assert_called_with(self.expectedTopic, self.expectedMessage)
p.kill()
except Exception as e:
p.kill()
raise Exception(e.args)
开发者ID:kki32，项目名称:microgearlib_test_python，代码行数:35，代码来源:all.py
示例10: test
def test():
gearkey = "qnlgzsPUUxYeyQP"
gearsecret = "1euJPvxybllEPQZzq2u9wpRJXDbjM7"
appid = "testNo3"
if(os.path.isfile("microgear.cache")):
f = open((os.getcwd() + "/microgear.cache"), 'r')
print(f.readlines())
f.close()
else:
print("yes1")
client.create(gearkey, gearsecret, appid, {'debugmode': "True", 'scope': "chat:receiver"})
client.setname("sender")
if(os.path.isfile("microgear.cache")):
f = open((os.getcwd() + "/microgear.cache"), 'r')
print(f.readlines())
f.close()
else:
print("yes2")
client.connect()
f = open((os.getcwd() + "/microgear.cache"), 'r')
print(f.readlines())
f.close()
client.resettoken()
if(os.path.isfile("microgear.cache")):
f = open((os.getcwd() + "/microgear.cache"), 'r')
print(f.readlines())
f.close()
else:
print("yes3")
开发者ID:kki32，项目名称:MicroGearTestingSuite，代码行数:30，代码来源:testScript.py
示例11: testCode4Case2
def testCode4Case2(self):
"""chat with other microgear in same appid"""
try:
print('Code4Case2')
print("run helper...")
code = str(31)
args = ['python', 'helper.py', code]
p = subprocess.Popen(args, cwd=(helper_dir))
time.sleep(connect_worst_timeout)
print("run main...")
client.create(self.gearkey, self.gearsecret, self.appid)
client.setalias(self.gearname)
client.on_connect = MagicMock()
client.connect()
time.sleep(connect_timeout)
self.assertTrue(client.on_connect.called)
self.connected = True
client.chat(self.helperGearname, self.message)
time.sleep(message_timeout)
r_file = open(receiver_file, "r")
received_message = r_file.read()
r_file.close()
if(received_message == self.message):
self.received = True
self.assertTrue(self.received)
p.kill()
#if fails due to assertion error
except Exception as e:
print("fail")
raise Exception(e.args)
开发者ID:kki32，项目名称:microgearlib_test_python，代码行数:34，代码来源:all.py
示例12: testAlias
def testAlias():
gearkey = "qnlgzsPUUxYeyQP"
gearsecret = "1euJPvxybllEPQZzq2u9wpRJXDbjM7"
appid = "testNo3"
client.create(gearkey, gearsecret, appid, {'debugmode': "True", 'alias': "Python"})
client.connect()
while True:
pass
开发者ID:kki32，项目名称:MicroGearTestingSuite，代码行数:9，代码来源:testScript.py
示例13: testConnectWillBlockFalse
def testConnectWillBlockFalse(self):
self.connected = False
def on_connected():
self.connected = True
client.on_connect = on_connected
client.create(self.gearkey, self.gearsecret, self.appid, {'debugmode': "True"})
client.connect(False)
timeout = time.time() + 30.0
if(time.time() > timeout or self.connected):
self.assertTrue(self.connected)
开发者ID:kki32，项目名称:MicroGearTestingSuite，代码行数:10，代码来源:testConnect.py
示例14: testCreateNetPieConnection
def testCreateNetPieConnection():
gearkey, gearsecret, appid = testCreateNetPie1();
client.create(gearkey, gearsecret, appid, {'debugmode' : False})
def on_connection():
print "I am connected"
client.on_connect = on_connection
print(client.on_connect == 0)
client.connect()
print("Sleep for 100 seconds")
time.sleep(100)
开发者ID:kki32，项目名称:MicroGearTestingSuite，代码行数:11，代码来源:testCreate.py
示例15: testCreateNetPieScopeChat
def testCreateNetPieScopeChat():
gearkey, gearsecret, appid = testCreateNetPie1();
client.create(gearkey, gearsecret, appid, {'debugmode' : True, 'scope' : "chat:java ja"})
client.setname('Python ja')
client.connect()
def receive_message(topic, message):
print topic + " " + message
while(True):
client.chat('Html ka', "Hello html")
time.sleep(3)
client.on_message = receive_message
开发者ID:kki32，项目名称:MicroGearTestingSuite，代码行数:13，代码来源:testCreate.py
示例16: testChat
def testChat():
gearkey = "ExhoyeQoTyJS5Ac"
gearsecret = "gzDawaaHRe1KvQhepAw3WYuuGHjBsh"
appid = "p107microgear"
origin = "oriA"
destination = "destX"
client.create(gearkey , gearsecret, appid, {'debugmode': True})
bf = open(os.path.join(os.getcwd(),"microgear.cache"), "rb")
print(bf)
client.resettoken()
af = open(os.path.join(os.getcwd(),"microgear.cache"), "rb")
print(af)
开发者ID:kki32，项目名称:microgearlib_test_python，代码行数:14，代码来源:testChatWithMyself.py
示例17: testScopeChat
def testScopeChat():
gearkey = "qnlgzsPUUxYeyQP"
gearsecret = "1euJPvxybllEPQZzq2u9wpRJXDbjM7"
appid = "testNo3"
client.create(gearkey, gearsecret, appid, {'debugmode': "True", 'scope': "chat:receiver"})
client.setname("sender")
client.connect()
def receive_message(topic, message):
print topic + " " + message
while True:
client.chat("not receiver","How are you?")
time.sleep(3)
开发者ID:kki32，项目名称:MicroGearTestingSuite，代码行数:15，代码来源:testScript.py
示例18: testCreateNetPieScopeW
def testCreateNetPieScopeW():
gearkey = "ExhoyeQoTyJS5Ac"
gearsecret = "gzDawaaHRe1KvQhepAw3WYuuGHjBsh"
appid = "p107microgear"
client.create(gearkey , gearsecret, appid, {'debugmode': True,'scope': "r:/LetsShare" })
client.create(gearkey , gearsecret, appid, {'debugmode': True,'scope': "w:/LetsShare" })
client.setname("Python ja")
client.connect()
def receive_message(topic, message):
print topic + " " + message
while True:
client.publish("/StopsShare","Happy New Year!")
time.sleep(3)
client.on_message = receive_message
开发者ID:kki32，项目名称:MicroGearTestingSuite，代码行数:16，代码来源:testCreate.py
示例19: testCode4Case1
def testCode4Case1(self):
"""chat with itself"""
print('Code4Case1')
client.create(self.gearkey, self.gearsecret, self.appid)
client.setalias(self.gearname)
client.on_message = MagicMock()
client.on_connect = MagicMock()
client.connect()
time.sleep(connect_timeout)
self.assertTrue(client.on_connect.called)
self.connected = True
client.chat(self.gearname, self.message)
time.sleep(message_timeout)
self.assertTrue(client.on_message.called)
client.on_message.assert_called_once_with(self.expectedMsgTopic, self.expectedMessage)
开发者ID:kki32，项目名称:microgearlib_test_python，代码行数:16，代码来源:all.py
示例20: testCode8Case1
def testCode8Case1(self):
"""resettoken when no microgear.cache
pre-requisite: no microgear.cache file"""
if(os.path.isfile(microgear_cache)):
os.remove(microgear_cache)
self.assertIsNone(microgear.gearkey)
self.assertIsNone(microgear.gearsecret)
self.assertIsNone(microgear.appid)
client.create(self.gearkey, self.gearsecret, self.appid)
client.on_connect = MagicMock()
client.resettoken()
self.assertFalse(os.path.isfile(microgear_cache))
开发者ID:kki32，项目名称:microgearlib_test_python，代码行数:16，代码来源:unitTestPython.py
注：本文中的microgear.client.create函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论