diff options
Diffstat (limited to 'python/tests')
| -rwxr-xr-x | python/tests/subscribe-test.py | 154 | ||||
| -rw-r--r-- | python/tests/test_grant_async.py | 369 | ||||
| -rw-r--r-- | python/tests/test_history.py | 37 | ||||
| -rw-r--r-- | python/tests/test_publish_async.py | 228 |
4 files changed, 37 insertions, 751 deletions
diff --git a/python/tests/subscribe-test.py b/python/tests/subscribe-test.py deleted file mode 100755 index 7a4b723..0000000 --- a/python/tests/subscribe-test.py +++ /dev/null @@ -1,154 +0,0 @@ -## www.pubnub.com - PubNub Real-time push service in the cloud. -# coding=utf8 - -## PubNub Real-time Push APIs and Notifications Framework -## Copyright (c) 2010 Stephen Blum -## http://www.pubnub.com/ - -## ----------------------------------- -## PubNub 3.1 Real-time Push Cloud API -## ----------------------------------- - -import sys -import datetime -from pubnub import PubnubAsync as Pubnub -from functools import partial -from threading import current_thread -import threading -publish_key = len(sys.argv) > 1 and sys.argv[1] or 'demo' -subscribe_key = len(sys.argv) > 2 and sys.argv[2] or 'demo' -secret_key = len(sys.argv) > 3 and sys.argv[3] or 'demo' -cipher_key = len(sys.argv) > 4 and sys.argv[4] or None -ssl_on = len(sys.argv) > 5 and bool(sys.argv[5]) or False - -## ----------------------------------------------------------------------- -## Initiate Pubnub State -## ----------------------------------------------------------------------- -#pubnub = Pubnub( publish_key, subscribe_key, secret_key, cipher_key, ssl_on ) -pubnub = Pubnub(publish_key, subscribe_key, secret_key, ssl_on) -crazy = 'hello_world' - -current = -1 - -errors = 0 -received = 0 - -## ----------------------------------------------------------------------- -## Subscribe Example -## ----------------------------------------------------------------------- - - -def message_received(message): - print(message) - - -def check_received(message): - global current - global errors - global received - print(message) - print(current) - if message <= current: - print('ERROR') - #sys.exit() - errors += 1 - else: - received += 1 - print('active thread count : ' + str(threading.activeCount())) - print('errors = ' + str(errors)) - print(current_thread().getName() + ' , ' + 'received = ' + str(received)) - - if received != message: - print('********** MISSED **************** ' + str(message - received)) - current = message - - -def connected_test(ch): - print('Connected ' + ch) - - -def connected(ch): - pass - - -''' -pubnub.subscribe({ - 'channel' : 'abcd1', - 'connect' : connected, - 'callback' : message_received -}) -''' - - -def cb1(): - pubnub.subscribe({ - 'channel': 'efgh1', - 'connect': connected, - 'callback': message_received - }) - - -def cb2(): - pubnub.subscribe({ - 'channel': 'dsm-test', - 'connect': connected_test, - 'callback': check_received - }) - - -def cb3(): - pubnub.unsubscribe({'channel': 'efgh1'}) - - -def cb4(): - pubnub.unsubscribe({'channel': 'abcd1'}) - - -def subscribe(channel): - pubnub.subscribe({ - 'channel': channel, - 'connect': connected, - 'callback': message_received - }) - - -pubnub.timeout(15, cb1) - -pubnub.timeout(30, cb2) - - -pubnub.timeout(45, cb3) - -pubnub.timeout(60, cb4) - -#''' -for x in range(1, 1000): - #print x - def y(t): - subscribe('channel-' + str(t)) - - def z(t): - pubnub.unsubscribe({'channel': 'channel-' + str(t)}) - - pubnub.timeout(x + 5, partial(y, x)) - pubnub.timeout(x + 25, partial(z, x)) - x += 10 -#''' - -''' -for x in range(1,1000): - def cb(r): print r , ' : ', threading.activeCount() - def y(t): - pubnub.publish({ - 'message' : t, - 'callback' : cb, - 'channel' : 'dsm-test' - }) - - - pubnub.timeout(x + 1, partial(y,x)) - x += 1 -''' - - -pubnub.start() diff --git a/python/tests/test_grant_async.py b/python/tests/test_grant_async.py deleted file mode 100644 index 07c8a2c..0000000 --- a/python/tests/test_grant_async.py +++ /dev/null @@ -1,369 +0,0 @@ - - -from pubnub import Pubnub -import time - -pubnub = Pubnub("demo","demo") -pubnub_pam = Pubnub("pub-c-c077418d-f83c-4860-b213-2f6c77bde29a", - "sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe", "sec-c-OGU3Y2Q4ZWUtNDQwMC00NTI1LThjNWYtNWJmY2M4OGIwNjEy") - - - -# Grant permission read true, write true, on channel ( Async Mode ) -def test_1(): - resp = {'response' : None} - def _callback(response): - resp['response'] = response - - def _error(response): - resp['response'] = response - - pubnub_pam.grant(channel="abcd", auth_key="abcd", read=True, write=True, ttl=1, callback=_callback, error=_error) - time.sleep(2) - assert resp['response'] == { - 'message': u'Success', - 'payload': {u'auths': {u'abcd': {u'r': 1, u'w': 1}}, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'user', u'channel': u'abcd', u'ttl': 1} - } - - -# Grant permission read false, write false, on channel ( Async Mode ) -def test_2(): - resp = {'response' : None} - def _callback(response): - resp['response'] = response - - def _error(response): - resp['response'] = response - - pubnub_pam.grant(channel="abcd", auth_key="abcd", read=False, write=False, ttl=1, callback=_callback, error=_error) - time.sleep(2) - assert resp['response'] == { - 'message': u'Success', - 'payload': {u'auths': {u'abcd': {u'r': 0, u'w': 0}}, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'user', u'channel': u'abcd', u'ttl': 1} - } - - -# Grant permission read True, write false, on channel ( Async Mode ) -def test_3(): - resp = {'response' : None} - def _callback(response): - resp['response'] = response - - def _error(response): - resp['response'] = response - - pubnub_pam.grant(channel="abcd", auth_key="abcd", read=True, write=False, ttl=1, callback=_callback, error=_error) - time.sleep(2) - assert resp['response'] == { - 'message': u'Success', - 'payload': {u'auths': {u'abcd': {u'r': 1, u'w': 0}}, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'user', u'channel': u'abcd', u'ttl': 1} - } - -# Grant permission read False, write True, on channel ( Async Mode ) -def test_4(): - resp = {'response' : None} - def _callback(response): - resp['response'] = response - - def _error(response): - resp['response'] = response - - pubnub_pam.grant(channel="abcd", auth_key="abcd", read=False, write=True, ttl=1, callback=_callback, error=_error) - time.sleep(2) - assert resp['response'] == { - 'message': u'Success', - 'payload': {u'auths': {u'abcd': {u'r': 0, u'w': 1}}, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'user', u'channel': u'abcd', u'ttl': 1} - } - - -# Grant permission read False, write True, on channel ( Async Mode ), TTL 10 -def test_5(): - resp = {'response' : None} - def _callback(response): - resp['response'] = response - - def _error(response): - resp['response'] = response - - pubnub_pam.grant(channel="abcd", auth_key="abcd", read=False, write=True, ttl=10, callback=_callback, error=_error) - time.sleep(2) - assert resp['response'] == { - 'message': u'Success', - 'payload': {u'auths': {u'abcd': {u'r': 0, u'w': 1}}, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'user', u'channel': u'abcd', u'ttl': 10} - } - - -# Grant permission read False, write True, without channel ( Async Mode ), TTL 10 -def test_6(): - resp = {'response' : None} - def _callback(response): - resp['response'] = response - - def _error(response): - resp['response'] = response - - pubnub_pam.grant(auth_key="abcd", read=False, write=True, ttl=10, callback=_callback, error=_error) - time.sleep(2) - assert resp['response'] == { - 'message': u'Success', - 'payload': { u'r': 0, u'w': 1, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'subkey', u'ttl': 10} - } - - -# Grant permission read False, write False, without channel ( Async Mode ) -def test_7(): - resp = {'response' : None} - def _callback(response): - resp['response'] = response - - def _error(response): - resp['response'] = response - - pubnub_pam.grant(auth_key="abcd", read=False, write=False, callback=_callback, error=_error) - time.sleep(2) - assert resp['response'] == { - 'message': u'Success', - 'payload': { u'r': 0, u'w': 0, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'subkey', u'ttl': 1} - } - - -# Complete flow , try publish on forbidden channel, grant permission to subkey and try again. ( Sync Mode) - -def test_8(): - channel = "test_8-" + str(time.time()) - message = "Hello World" - auth_key = "auth-" + channel - pubnub_pam.set_auth_key(auth_key) - - - - def _cb1(resp, ch=None): - assert False - def _err1(resp): - assert resp['message'] == 'Forbidden' - assert resp['payload'] == {u'channels': [channel]} - pubnub_pam.publish(channel=channel,message=message, callback=_cb1, error=_err1) - time.sleep(2) - - - def _cb2(resp, ch=None): - assert resp == { - 'message': u'Success', - 'payload': {u'auths': {auth_key : {u'r': 1, u'w': 1}}, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'user', u'channel': channel, u'ttl': 10} - } - def _err2(resp): - assert False - - - resp = pubnub_pam.grant(channel=channel, read=True, write=True, auth_key=auth_key, ttl=10, callback=_cb2, error=_err2) - time.sleep(2) - - def _cb3(resp, ch=None): - assert resp[0] == 1 - def _err3(resp): - assert False - - pubnub_pam.publish(channel=channel,message=message, callback=_cb3, error=_err3) - time.sleep(2) - - - - - -# Complete flow , try publish on forbidden channel, grant permission to authkey and try again. -# then revoke and try again -def test_9(): - channel = "test_9-" + str(time.time()) - message = "Hello World" - auth_key = "auth-" + channel - pubnub_pam.set_auth_key(auth_key) - - def _cb1(resp, ch=None): - assert False - def _err1(resp): - assert resp['message'] == 'Forbidden' - assert resp['payload'] == {u'channels': [channel]} - pubnub_pam.publish(channel=channel,message=message, callback=_cb1, error=_err1) - time.sleep(2) - - - def _cb2(resp, ch=None): - assert resp == { - 'message': u'Success', - 'payload': {u'auths': {auth_key : {u'r': 1, u'w': 1}}, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'user', u'channel': channel, u'ttl': 10} - } - def _err2(resp): - assert False - - - pubnub_pam.grant(channel=channel, read=True, write=True, auth_key=auth_key, ttl=10, callback=_cb2, error=_err2) - time.sleep(2) - - def _cb3(resp, ch=None): - assert resp[0] == 1 - def _err3(resp): - assert False - - pubnub_pam.publish(channel=channel,message=message, callback=_cb3, error=_err3) - time.sleep(2) - - def _cb4(resp, ch=None): - assert resp == { - 'message': u'Success', - 'payload': {u'auths': {auth_key : {u'r': 0, u'w': 0}}, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'user', u'channel': channel, u'ttl': 1} - } - def _err4(resp): - assert False - pubnub_pam.revoke(channel=channel, auth_key=auth_key, callback=_cb4, error=_err4) - time.sleep(2) - - def _cb5(resp, ch=None): - assert False - def _err5(resp): - assert resp['message'] == 'Forbidden' - assert resp['payload'] == {u'channels': [channel]} - pubnub_pam.publish(channel=channel,message=message, callback=_cb5, error=_err5) - time.sleep(2) - - - - -# Complete flow , try publish on forbidden channel, grant permission channel level for subkey and try again. -# then revoke and try again -def test_10(): - channel = "test_10-" + str(time.time()) - message = "Hello World" - auth_key = "auth-" + channel - pubnub_pam.set_auth_key(auth_key) - - def _cb1(resp, ch=None): - assert False - def _err1(resp): - assert resp['message'] == 'Forbidden' - assert resp['payload'] == {u'channels': [channel]} - pubnub_pam.publish(channel=channel,message=message, callback=_cb1, error=_err1) - time.sleep(2) - - - def _cb2(resp, ch=None): - assert resp == { - 'message': u'Success', - 'payload': { u'channels': {channel: {u'r': 1, u'w': 1}}, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'channel', u'ttl': 10} - } - def _err2(resp): - assert False - - - pubnub_pam.grant(channel=channel, read=True, write=True, ttl=10, callback=_cb2, error=_err2) - time.sleep(2) - - def _cb3(resp, ch=None): - assert resp[0] == 1 - def _err3(resp): - assert False - - pubnub_pam.publish(channel=channel,message=message, callback=_cb3, error=_err3) - time.sleep(2) - - def _cb4(resp, ch=None): - assert resp == { - 'message': u'Success', - 'payload': { u'channels': {channel : {u'r': 0, u'w': 0}}, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'channel', u'ttl': 1} - } - def _err4(resp): - assert False - pubnub_pam.revoke(channel=channel, callback=_cb4, error=_err4) - time.sleep(2) - - def _cb5(resp, ch=None): - assert False - def _err5(resp): - assert resp['message'] == 'Forbidden' - assert resp['payload'] == {u'channels': [channel]} - pubnub_pam.publish(channel=channel,message=message, callback=_cb5, error=_err5) - time.sleep(2) - - - -# Complete flow , try publish on forbidden channel, grant permission subkey level for subkey and try again. -# then revoke and try again -def test_11(): - channel = "test_11-" + str(time.time()) - message = "Hello World" - auth_key = "auth-" + channel - pubnub_pam.set_auth_key(auth_key) - - def _cb1(resp, ch=None): - assert False - def _err1(resp): - assert resp['message'] == 'Forbidden' - assert resp['payload'] == {u'channels': [channel]} - pubnub_pam.publish(channel=channel,message=message, callback=_cb1, error=_err1) - time.sleep(2) - - - def _cb2(resp, ch=None): - assert resp == { - 'message': u'Success', - 'payload': { u'r': 1, u'w': 1, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'subkey', u'ttl': 10} - } - def _err2(resp): - assert False - - - pubnub_pam.grant(read=True, write=True, ttl=10, callback=_cb2, error=_err2) - time.sleep(2) - - def _cb3(resp, ch=None): - assert resp[0] == 1 - def _err3(resp): - assert False - - pubnub_pam.publish(channel=channel,message=message, callback=_cb3, error=_err3) - time.sleep(2) - - def _cb4(resp, ch=None): - assert resp == { - 'message': u'Success', - 'payload': {u'r': 0, u'w': 0, - u'subscribe_key': u'sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe', - u'level': u'subkey', u'ttl': 1} - } - def _err4(resp): - assert False - pubnub_pam.revoke(callback=_cb4, error=_err4) - time.sleep(2) - - def _cb5(resp, ch=None): - assert False - def _err5(resp): - assert resp['message'] == 'Forbidden' - assert resp['payload'] == {u'channels': [channel]} - pubnub_pam.publish(channel=channel,message=message, callback=_cb5, error=_err5) - time.sleep(2) diff --git a/python/tests/test_history.py b/python/tests/test_history.py new file mode 100644 index 0000000..3cf8e03 --- /dev/null +++ b/python/tests/test_history.py @@ -0,0 +1,37 @@ +from pubnub import Pubnub +import time +import random +from nose.tools import with_setup + + +pubnub = Pubnub("ds","ds") +pubnub_enc = Pubnub(publish_key="ds",subscribe_key="ds",cipher_key="enigma") + + +def rand(msg): + return "rand-" + str(random.random()) + "-" + msg + +channel = rand("channel") +channel_enc = rand("channel_enc") + +messages = [] + + +def setup_func(): + for i in range(0,20): + msg = rand("message-" + str(i)) + messages.append(msg) + pubnub.publish(channel=channel, message=msg) + pubnub_enc.publish(channel=channel_enc, message=msg) + + +@with_setup(setup_func) +def test_1(): + time.sleep(3) + hresp = pubnub.history(channel=channel, count=20) + hresp2 = pubnub_enc.history(channel=channel_enc, count=20) + assert hresp[0] == messages + assert hresp2[0] == messages + + + diff --git a/python/tests/test_publish_async.py b/python/tests/test_publish_async.py deleted file mode 100644 index cf0811e..0000000 --- a/python/tests/test_publish_async.py +++ /dev/null @@ -1,228 +0,0 @@ - - -from pubnub import Pubnub -import time - -pubnub = Pubnub("demo","demo") -pubnub_enc = Pubnub("demo", "demo", cipher_key="enigma") -pubnub_pam = Pubnub("pub-c-c077418d-f83c-4860-b213-2f6c77bde29a", - "sub-c-e8839098-f568-11e2-a11a-02ee2ddab7fe", "sec-c-OGU3Y2Q4ZWUtNDQwMC00NTI1LThjNWYtNWJmY2M4OGIwNjEy") - - - -# Publish and receive string -def test_1(): - - channel = "test_1-" + str(time.time()) - message = "I am a string" - - def _cb(resp, ch=None): - assert resp == message - pubnub.unsubscribe(channel) - - def _connect(resp): - pubnub.publish(channel,message) - - def _error(resp): - assert False - - pubnub.subscribe(channel, callback=_cb, connect=_connect, error=_error) - -# Publish and receive array -def test_2(): - - channel = "test_2-" + str(time.time()) - message = [1,2] - - def _cb(resp, ch=None): - assert resp == message - pubnub.unsubscribe(channel) - - def _connect(resp): - pubnub.publish(channel,message) - - def _error(resp): - assert False - - pubnub.subscribe(channel, callback=_cb, connect=_connect, error=_error) - -# Publish and receive json object -def test_3(): - - channel = "test_2-" + str(time.time()) - message = { "a" : "b" } - - def _cb(resp, ch=None): - assert resp == message - pubnub.unsubscribe(channel) - - def _connect(resp): - pubnub.publish(channel,message) - - def _error(resp): - assert False - - pubnub.subscribe(channel, callback=_cb, connect=_connect, error=_error) - -# Publish and receive number -def test_4(): - - channel = "test_2-" + str(time.time()) - message = 100 - - def _cb(resp, ch=None): - assert resp == message - pubnub.unsubscribe(channel) - - def _connect(resp): - pubnub.publish(channel,message) - - def _error(resp): - assert False - - pubnub.subscribe(channel, callback=_cb, connect=_connect, error=_error) - -# Publish and receive number string -def test_5(): - - channel = "test_5-" + str(time.time()) - message = "100" - - def _cb(resp, ch=None): - assert resp == message - pubnub.unsubscribe(channel) - - def _connect(resp): - pubnub.publish(channel,message) - - def _error(resp): - assert False - - pubnub.subscribe(channel, callback=_cb, connect=_connect, error=_error) - - -# Publish and receive string (Encryption enabled) -def test_6(): - - channel = "test_6-" + str(time.time()) - message = "I am a string" - - def _cb(resp, ch=None): - assert resp == message - pubnub_enc.unsubscribe(channel) - - def _connect(resp): - pubnub_enc.publish(channel,message) - - def _error(resp): - assert False - - pubnub_enc.subscribe(channel, callback=_cb, connect=_connect, error=_error) - -# Publish and receive array (Encryption enabled) -def test_7(): - - channel = "test_7-" + str(time.time()) - message = [1,2] - - def _cb(resp, ch=None): - assert resp == message - pubnub_enc.unsubscribe(channel) - - def _connect(resp): - pubnub.publish(channel,message) - - def _error(resp): - assert False - - pubnub.subscribe(channel, callback=_cb, connect=_connect, error=_error) - -# Publish and receive json object (Encryption enabled) -def test_8(): - - channel = "test_8-" + str(time.time()) - message = { "a" : "b" } - - def _cb(resp, ch=None): - assert resp == message - pubnub_enc.unsubscribe(channel) - - def _connect(resp): - pubnub_enc.publish(channel,message) - - def _error(resp): - assert False - - pubnub_enc.subscribe(channel, callback=_cb, connect=_connect, error=_error) - -# Publish and receive number (Encryption enabled) -def test_9(): - - channel = "test_9-" + str(time.time()) - message = 100 - - def _cb(resp, ch=None): - assert resp == message - pubnub_enc.unsubscribe(channel) - - def _connect(resp): - pubnub_enc.publish(channel,message) - - def _error(resp): - assert False - - pubnub_enc.subscribe(channel, callback=_cb, connect=_connect, error=_error) - -# Publish and receive number string (Encryption enabled) -def test_10(): - - channel = "test_10-" + str(time.time()) - message = "100" - - def _cb(resp, ch=None): - assert resp == message - pubnub_enc.unsubscribe(channel) - - def _connect(resp): - pubnub_enc.publish(channel,message) - - def _error(resp): - assert False - - pubnub_enc.subscribe(channel, callback=_cb, connect=_connect, error=_error) - -# Publish and receive object string (Encryption enabled) -def test_11(): - - channel = "test_11-" + str(time.time()) - message = '{"a" : "b"}' - - def _cb(resp, ch=None): - assert resp == message - pubnub_enc.unsubscribe(channel) - - def _connect(resp): - pubnub_enc.publish(channel,message) - - def _error(resp): - assert False - - pubnub_enc.subscribe(channel, callback=_cb, connect=_connect, error=_error) - -# Publish and receive array string (Encryption enabled) -def test_12(): - - channel = "test_12-" + str(time.time()) - message = '[1,2]' - - def _cb(resp, ch=None): - assert resp == message - pubnub_enc.unsubscribe(channel) - - def _connect(resp): - pubnub_enc.publish(channel,message) - - def _error(resp): - assert False - - pubnub_enc.subscribe(channel, callback=_cb, connect=_connect, error=_error) |