aboutsummaryrefslogtreecommitdiffstats
path: root/python-tornado/tests
diff options
context:
space:
mode:
Diffstat (limited to 'python-tornado/tests')
-rw-r--r--python-tornado/tests/benchmark.py96
-rw-r--r--python-tornado/tests/delivery.py161
-rw-r--r--python-tornado/tests/unit-test.py224
3 files changed, 481 insertions, 0 deletions
diff --git a/python-tornado/tests/benchmark.py b/python-tornado/tests/benchmark.py
new file mode 100644
index 0000000..5f9b5e3
--- /dev/null
+++ b/python-tornado/tests/benchmark.py
@@ -0,0 +1,96 @@
+## 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
+import tornado
+sys.path.append('../')
+from Pubnub import Pubnub
+
+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 'demo'
+ssl_on = len(sys.argv) > 5 and bool(sys.argv[5]) or False
+origin = len(sys.argv) > 6 and sys.argv[6] or 'pubsub.pubnub.com'
+
+
+## -----------------------------------------------------------------------
+## Initiat Class
+## -----------------------------------------------------------------------
+pubnub = Pubnub(
+ publish_key,
+ subscribe_key,
+ secret_key = secret_key,
+ cipher_key = cipher_key,
+ ssl_on = ssl_on,
+ origin = origin
+)
+crazy = ' ~`!@#$%^&*( 顶顅 Ȓ)+=[]\\{}|;\':",./<>?abcd'
+
+## -----------------------------------------------------------------------
+## BENCHMARK
+## -----------------------------------------------------------------------
+def connected() :
+ pubnub.publish({
+ 'channel' : crazy,
+ 'message' : { 'Info' : 'Connected!' }
+ })
+
+trips = { 'last' : None, 'current' : None, 'max' : 0, 'avg' : 0 }
+
+def received(message):
+ current_trip = trips['current'] = str(datetime.datetime.now())[0:19]
+ last_trip = trips['last'] = str(
+ datetime.datetime.now() - datetime.timedelta(seconds=1)
+ )[0:19]
+
+ ## New Trip Span (1 Second)
+ if not trips.has_key(current_trip) :
+ trips[current_trip] = 0
+
+ ## Average
+ if trips.has_key(last_trip):
+ trips['avg'] = (trips['avg'] + trips[last_trip]) / 2
+
+ ## Increment Trip Counter
+ trips[current_trip] = trips[current_trip] + 1
+
+ ## Update Max
+ if trips[current_trip] > trips['max'] :
+ trips['max'] = trips[current_trip]
+
+
+ print(message)
+
+ pubnub.publish({
+ 'channel' : crazy,
+ 'message' : current_trip +
+ " Trip: " +
+ str(trips[current_trip]) +
+ " MAX: " +
+ str(trips['max']) +
+ "/sec " +
+ " AVG: " +
+ str(trips['avg']) +
+ "/sec"
+ })
+
+pubnub.subscribe({
+ 'channel' : crazy,
+ 'connect' : connected,
+ 'callback' : received
+})
+
+## -----------------------------------------------------------------------
+## IO Event Loop
+## -----------------------------------------------------------------------
+pubnub.start()
diff --git a/python-tornado/tests/delivery.py b/python-tornado/tests/delivery.py
new file mode 100644
index 0000000..f3633e6
--- /dev/null
+++ b/python-tornado/tests/delivery.py
@@ -0,0 +1,161 @@
+## 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
+import time
+import math
+
+sys.path.append('../')
+from Pubnub import Pubnub
+
+## -----------------------------------------------------------------------
+## Configuration
+## -----------------------------------------------------------------------
+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 'demo'
+ssl_on = len(sys.argv) > 5 and bool(sys.argv[5]) or False
+origin = len(sys.argv) > 6 and sys.argv[6] or 'pubsub.pubnub.com'
+
+## -----------------------------------------------------------------------
+## Analytics
+## -----------------------------------------------------------------------
+analytics = {
+ 'publishes' : 0, ## Total Send Requests
+ 'received' : 0, ## Total Received Messages (Deliveries)
+ 'queued' : 0, ## Total Unreceived Queue (UnDeliveries)
+ 'successful_publishes' : 0, ## Confirmed Successful Publish Request
+ 'failed_publishes' : 0, ## Confirmed UNSuccessful Publish Request
+ 'failed_deliveries' : 0, ## (successful_publishes - received)
+ 'deliverability' : 0 ## Percentage Delivery
+}
+
+trips = {
+ 'last' : None,
+ 'current' : None,
+ 'max' : 0,
+ 'avg' : 0
+}
+
+## -----------------------------------------------------------------------
+## Initiat Class
+## -----------------------------------------------------------------------
+channel = 'deliverability-' + str(time.time())
+pubnub = Pubnub(
+ publish_key,
+ subscribe_key,
+ secret_key = secret_key,
+ cipher_key = cipher_key,
+ ssl_on = ssl_on,
+ origin = origin
+)
+
+## -----------------------------------------------------------------------
+## BENCHMARK
+## -----------------------------------------------------------------------
+def publish_sent(info = None):
+ if info and info[0]: analytics['successful_publishes'] += 1
+ else: analytics['failed_publishes'] += 1
+
+ analytics['publishes'] += 1
+ analytics['queued'] += 1
+
+ pubnub.timeout( send, 0.1 )
+
+def send():
+ if analytics['queued'] > 100:
+ analytics['queued'] -= 10
+ return pubnub.timeout( send, 10 )
+
+ pubnub.publish({
+ 'channel' : channel,
+ 'callback' : publish_sent,
+ 'message' : "1234567890"
+ })
+
+def received(message):
+ analytics['queued'] -= 1
+ analytics['received'] += 1
+ current_trip = trips['current'] = str(datetime.datetime.now())[0:19]
+ last_trip = trips['last'] = str(
+ datetime.datetime.now() - datetime.timedelta(seconds=1)
+ )[0:19]
+
+ ## New Trip Span (1 Second)
+ if not trips.has_key(current_trip) :
+ trips[current_trip] = 0
+
+ ## Average
+ if trips.has_key(last_trip):
+ trips['avg'] = (trips['avg'] + trips[last_trip]) / 2
+
+ ## Increment Trip Counter
+ trips[current_trip] = trips[current_trip] + 1
+
+ ## Update Max
+ if trips[current_trip] > trips['max'] :
+ trips['max'] = trips[current_trip]
+
+def show_status():
+ ## Update Failed Deliveries
+ analytics['failed_deliveries'] = \
+ analytics['successful_publishes'] \
+ - analytics['received']
+
+ ## Update Deliverability
+ analytics['deliverability'] = (
+ float(analytics['received']) / \
+ float(analytics['successful_publishes'] or 1.0)
+ ) * 100.0
+
+ ## Print Display
+ print( (
+ "max:%(max)03d/sec " + \
+ "avg:%(avg)03d/sec " + \
+ "pubs:%(publishes)05d " + \
+ "received:%(received)05d " + \
+ "spub:%(successful_publishes)05d " + \
+ "fpub:%(failed_publishes)05d " + \
+ "failed:%(failed_deliveries)05d " + \
+ "queued:%(queued)03d " + \
+ "delivery:%(deliverability)03f%% " + \
+ ""
+ ) % {
+ 'max' : trips['max'],
+ 'avg' : trips['avg'],
+ 'publishes' : analytics['publishes'],
+ 'received' : analytics['received'],
+ 'successful_publishes' : analytics['successful_publishes'],
+ 'failed_publishes' : analytics['failed_publishes'],
+ 'failed_deliveries' : analytics['failed_deliveries'],
+ 'publishes' : analytics['publishes'],
+ 'deliverability' : analytics['deliverability'],
+ 'queued' : analytics['queued']
+ } )
+ pubnub.timeout( show_status, 1 )
+
+def connected():
+ show_status()
+ pubnub.timeout( send, 1 )
+
+print( "Connected: %s\n" % origin )
+pubnub.subscribe({
+ 'channel' : channel,
+ 'connect' : connected,
+ 'callback' : received
+})
+
+## -----------------------------------------------------------------------
+## IO Event Loop
+## -----------------------------------------------------------------------
+pubnub.start()
diff --git a/python-tornado/tests/unit-test.py b/python-tornado/tests/unit-test.py
new file mode 100644
index 0000000..f593d11
--- /dev/null
+++ b/python-tornado/tests/unit-test.py
@@ -0,0 +1,224 @@
+## 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/
+
+## TODO Tests
+##
+## - wait 20 minutes, send a message, receive and success.
+## -
+## -
+##
+##
+
+## -----------------------------------
+## PubNub 3.1 Real-time Push Cloud API
+## -----------------------------------
+
+import sys
+sys.path.append('../')
+from Pubnub import Pubnub
+
+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 None
+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
+
+## -----------------------------------------------------------------------
+## Command Line Options Supplied PubNub
+## -----------------------------------------------------------------------
+pubnub_user_supplied_options = Pubnub(
+ publish_key, ## OPTIONAL (supply None to disable)
+ subscribe_key, ## REQUIRED
+ secret_key, ## OPTIONAL (supply None to disable)
+ cipher_key, ## OPTIONAL (supply None to disable)
+ ssl_on ## OPTIONAL (supply None to disable)
+)
+
+## -----------------------------------------------------------------------
+## High Security PubNub
+## -----------------------------------------------------------------------
+pubnub_high_security = Pubnub(
+ ## Publish Key
+ 'pub-c-a30c030e-9f9c-408d-be89-d70b336ca7a0',
+
+ ## Subscribe Key
+ 'sub-c-387c90f3-c018-11e1-98c9-a5220e0555fd',
+
+ ## Secret Key
+ 'sec-c-MTliNDE0NTAtYjY4Ni00MDRkLTllYTItNDhiZGE0N2JlYzBl',
+
+ ## Cipher Key
+ 'YWxzamRmbVjFaa05HVnGFqZHM3NXRBS73jxmhVMkjiwVVXV1d5UrXR1JLSkZFRr'+
+ 'WVd4emFtUm1iR0TFpUZvbiBoYXMgYmVlbxWkhNaF3uUi8kM0YkJTEVlZYVFjBYi'+
+ 'jFkWFIxSkxTa1pGUjd874hjklaTFpUwRVuIFNob3VsZCB5UwRkxUR1J6YVhlQWa'+
+ 'V1ZkNGVH32mDkdho3pqtRnRVbTFpUjBaeGUgYXNrZWQtZFoKjda40ZWlyYWl1eX'+
+ 'U4RkNtdmNub2l1dHE2TTA1jd84jkdJTbFJXYkZwWlZtRnKkWVrSRhhWbFpZVmFz'+
+ 'c2RkZmTFpUpGa1dGSXhTa3hUYTFwR1Vpkm9yIGluZm9ybWFNfdsWQdSiiYXNWVX'+
+ 'RSblJWYlRGcFVqQmFlRmRyYUU0MFpXbHlZV2wxZVhVNFJrTnR51YjJsMWRIRTJU'+
+ 'W91ciBpbmZvcm1hdGliBzdWJtaXR0ZWQb3UZSBhIHJlc3BvbnNlLCB3ZWxsIHJl'+
+ 'VEExWdHVybiB0am0aW9uIb24gYXMgd2UgcG9zc2libHkgY2FuLuhcFe24ldWVns'+
+ 'dSaTFpU3hVUjFKNllWaFdhRmxZUWpCaQo34gcmVxdWlGFzIHNveqQl83snBfVl3',
+
+ ## 2048bit SSL ON - ENABLED TRUE
+ True
+)
+
+## -----------------------------------------------------------------------
+## Channel | Message Test Data (UTF-8)
+## -----------------------------------------------------------------------
+crazy = ' ~`â¦â§!@#$%^&*(顶顅Ȓ)+=[]\\{}|;\':",./<>?abcd'
+many_channels = [ str(x) + '-many_channel_test' for x in range(10) ]
+runthroughs = 0
+planned_tests = 2
+delivery_retries = 0
+max_retries = 10
+
+## -----------------------------------------------------------------------
+## Unit Test Function
+## -----------------------------------------------------------------------
+def test( trial, name ) :
+ if trial : print( 'PASS: ' + name )
+ else : print( '- FAIL - ' + name )
+
+def test_pubnub(pubnub):
+ global runthroughs, planned_tests, delivery_retries, max_retries
+
+ ## -----------------------------------------------------------------------
+ ## Many Channels
+ ## -----------------------------------------------------------------------
+ def phase2():
+ status = {
+ 'sent' : 0,
+ 'received' : 0,
+ 'connections' : 0
+ }
+
+ def received( message, chan ):
+ global runthroughs
+
+ test( status['received'] <= status['sent'], 'many sends' )
+ status['received'] += 1
+ pubnub.unsubscribe({ 'channel' : chan })
+ if status['received'] == len(many_channels):
+ runthroughs += 1
+ if runthroughs == planned_tests: pubnub.stop()
+
+ def publish_complete( info, chan ):
+ global delivery_retries, max_retries
+ status['sent'] += 1
+ test( info, 'publish complete' )
+ test( info and len(info) > 2, 'publish response' )
+ if not info[0]:
+ delivery_retries += 1
+ if max_retries > delivery_retries: sendit(chan)
+
+ def sendit(chan):
+ tchan = chan
+ pubnub.publish({
+ 'channel' : chan,
+ 'message' : "Hello World",
+ 'callback' : (lambda msg:publish_complete( msg, tchan ))
+ })
+
+ def connected(chan):
+ status['connections'] += 1
+ sendit(chan)
+
+ def delivered(info):
+ if info and info[0]: status['sent'] += 1
+
+ def subscribe(chan):
+ pubnub.subscribe({
+ 'channel' : chan,
+ 'connect' : (lambda:connected(chan+'')),
+ 'callback' : (lambda msg:received( msg, chan ))
+ })
+
+ ## Subscribe All Channels
+ for chan in many_channels: subscribe(chan)
+
+ ## -----------------------------------------------------------------------
+ ## Time Example
+ ## -----------------------------------------------------------------------
+ def time_complete(timetoken):
+ test( timetoken, 'timetoken fetch' )
+ test( isinstance( timetoken, int ), 'timetoken int type' )
+
+ pubnub.time({ 'callback' : time_complete })
+
+ ## -----------------------------------------------------------------------
+ ## Publish Example
+ ## -----------------------------------------------------------------------
+ def publish_complete(info):
+ test( info, 'publish complete' )
+ test( info and len(info) > 2, 'publish response' )
+
+ pubnub.history( {
+ 'channel' : crazy,
+ 'limit' : 10,
+ 'callback' : history_complete
+ })
+
+ ## -----------------------------------------------------------------------
+ ## History Example
+ ## -----------------------------------------------------------------------
+ def history_complete(messages):
+ test( messages and len(messages) > 0, 'history' )
+ test( messages, 'history' )
+
+
+ pubnub.publish({
+ 'channel' : crazy,
+ 'message' : "Hello World",
+ 'callback' : publish_complete
+ })
+
+ ## -----------------------------------------------------------------------
+ ## Subscribe Example
+ ## -----------------------------------------------------------------------
+ def message_received(message):
+ test( message, 'message received' )
+ pubnub.unsubscribe({ 'channel' : crazy })
+
+ def done() :
+ pubnub.unsubscribe({ 'channel' : crazy })
+ pubnub.publish({
+ 'channel' : crazy,
+ 'message' : "Hello World",
+ 'callback' : (lambda x:x)
+ })
+
+ def dumpster(message) :
+ test( 0, 'never see this' )
+
+ pubnub.subscribe({
+ 'channel' : crazy,
+ 'connect' : done,
+ 'callback' : dumpster
+ })
+
+ def connected() :
+ pubnub.publish({
+ 'channel' : crazy,
+ 'message' : { 'Info' : 'Connected!' }
+ })
+
+ pubnub.subscribe({
+ 'channel' : crazy,
+ 'connect' : connected,
+ 'callback' : message_received
+ })
+
+ phase2()
+
+## -----------------------------------------------------------------------
+## Run Tests
+## -----------------------------------------------------------------------
+test_pubnub(pubnub_user_supplied_options)
+test_pubnub(pubnub_high_security)
+pubnub_high_security.start()
+
generated by cgit v1.2.3 (git 2.25.1) at 2025-10-31 13:46:10 +0000