aboutsummaryrefslogtreecommitdiffstats
path: root/python/3.3
diff options
context:
space:
mode:
Diffstat (limited to 'python/3.3')
-rw-r--r--python/3.3/Pubnub.py57
-rwxr-xr-xpython/3.3/detailed-history-unit-test.py134
-rw-r--r--python/3.3/here-now-example.py33
-rwxr-xr-xpython/3.3/history-example.py12
-rwxr-xr-xpython/3.3/publish-example.py14
-rwxr-xr-xpython/3.3/subscribe-example.py64
-rwxr-xr-xpython/3.3/unit-test.py77
7 files changed, 0 insertions, 391 deletions
diff --git a/python/3.3/Pubnub.py b/python/3.3/Pubnub.py
deleted file mode 100644
index e806f9b..0000000
--- a/python/3.3/Pubnub.py
+++ /dev/null
@@ -1,57 +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.0 Real-time Push Cloud API
-## -----------------------------------
-
-try: import json
-except ImportError: import simplejson as json
-import sys
-import time
-import hashlib
-import urllib2
-import uuid
-from PubnubCore import PubnubCore
-
-class Pubnub(PubnubCore):
- def __init__(
- self,
- publish_key,
- subscribe_key,
- secret_key = False,
- cipher_key = False,
- ssl_on = False,
- origin = 'pubsub.pubnub.com',
- pres_uuid = None
- ) :
- super(Pubnub, self).__init__(
- publish_key = publish_key,
- subscribe_key = subscribe_key,
- secret_key = secret_key,
- cipher_key = cipher_key,
- ssl_on = ssl_on,
- origin = origin,
- uuid = pres_uuid
- )
-
- def _request( self, request, callback = None ) :
- ## Build URL
- url = self.getUrl(request)
-
- ## Send Request Expecting JSONP Response
- try:
- try: usock = urllib2.urlopen( url, None, 200 )
- except TypeError: usock = urllib2.urlopen( url, None )
- response = usock.read()
- usock.close()
- if (callback):
- callback(json.loads(response))
- else:
- return json.loads( response )
- except:
- return None
diff --git a/python/3.3/detailed-history-unit-test.py b/python/3.3/detailed-history-unit-test.py
deleted file mode 100755
index 2169e52..0000000
--- a/python/3.3/detailed-history-unit-test.py
+++ /dev/null
@@ -1,134 +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.0 Real-time Push Cloud API
-## -----------------------------------
-
-from Pubnub import Pubnub
-import unittest2 as unittest
-import sys
-
-
-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
-ssl_on = len(sys.argv) > 4 and bool(sys.argv[4]) or False
-pubnub = Pubnub(publish_key, subscribe_key, secret_key, ssl_on)
-crazy = ' ~`!@#$%^&*(顶顅Ȓ)+=[]\\{}|;\':",./<>?abcd'
-
-
-class TestDetailedHistory(unittest.TestCase):
- total_msg = 10
- channel = pubnub.time()
- starttime = None
- inputs = []
- endtime = None
- slice_a = 8
- slice_b = 2
- slice_size = slice_a - slice_b
-
- @classmethod
- def publish_msg(cls, start, end, offset):
- print 'Publishing messages'
- inputs = []
- for i in range(start + offset, end + offset):
- message = str(i) + " " + crazy
- success = pubnub.publish({
- 'channel': cls.channel,
- 'message': message,
- })
- t = pubnub.time()
- inputs.append({'timestamp': t, 'message': message})
- print 'Message # ', i, ' published'
- return inputs
-
- @classmethod
- def setUpClass(cls):
- print 'Setting up context for Detailed History tests. Please wait ...'
- cls.starttime = pubnub.time()
- cls.inputs = cls.inputs + cls.publish_msg(0, cls.total_msg / 2, 0)
- cls.midtime = pubnub.time()
- cls.inputs = cls.inputs + cls.publish_msg(
- 0, cls.total_msg / 2, cls.total_msg / 2)
- cls.endtime = pubnub.time()
- print 'Context setup for Detailed History tests. Now running tests'
-
- def test_begin_to_end_count(self):
- count = 5
- history = pubnub.detailedHistory({
- 'channel': self.__class__.channel,
- 'start': self.__class__.starttime,
- 'end': self.__class__.endtime,
- 'count': count
- })[0]
- self.assertTrue(len(history) == count and history[-1].encode(
- 'utf-8') == self.__class__.inputs[count - 1]['message'])
-
- def test_end_to_begin_count(self):
- count = 5
- history = pubnub.detailedHistory({
- 'channel': self.__class__.channel,
- 'start': self.__class__.endtime,
- 'end': self.__class__.starttime,
- 'count': count
- })[0]
- self.assertTrue(len(history) == count and history[-1]
- .encode('utf-8') == self.__class__.inputs[-1]['message'])
-
- def test_start_reverse_true(self):
- history = pubnub.detailedHistory({
- 'channel': self.__class__.channel,
- 'start': self.__class__.midtime,
- 'reverse': True
- })[0]
- self.assertTrue(len(history) == self.__class__.total_msg / 2)
- expected_msg = self.__class__.inputs[-1]['message']
- self.assertTrue(history[-1].encode('utf-8') == expected_msg)
-
- def test_start_reverse_false(self):
- history = pubnub.detailedHistory({
- 'channel': self.__class__.channel,
- 'start': self.__class__.midtime,
- })[0]
- self.assertTrue(history[0].encode('utf-8')
- == self.__class__.inputs[0]['message'])
-
- def test_end_reverse_true(self):
- history = pubnub.detailedHistory({
- 'channel': self.__class__.channel,
- 'end': self.__class__.midtime,
- 'reverse': True
- })[0]
- self.assertTrue(history[0].encode('utf-8')
- == self.__class__.inputs[0]['message'])
-
- def test_end_reverse_false(self):
- history = pubnub.detailedHistory({
- 'channel': self.__class__.channel,
- 'end': self.__class__.midtime,
- })[0]
- self.assertTrue(len(history) == self.__class__.total_msg / 2)
- self.assertTrue(history[-1].encode('utf-8')
- == self.__class__.inputs[-1]['message'])
-
- def test_count(self):
- history = pubnub.detailedHistory({
- 'channel': self.__class__.channel,
- 'count': 5
- })[0]
- self.assertTrue(len(history) == 5)
-
- def test_count_zero(self):
- history = pubnub.detailedHistory({
- 'channel': self.__class__.channel,
- 'count': 0
- })[0]
- self.assertTrue(len(history) == 0)
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/python/3.3/here-now-example.py b/python/3.3/here-now-example.py
deleted file mode 100644
index 261500a..0000000
--- a/python/3.3/here-now-example.py
+++ /dev/null
@@ -1,33 +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
-from twisted.internet import reactor
-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 ''
-ssl_on = len(sys.argv) > 5 and bool(sys.argv[5]) or False
-
-## -----------------------------------------------------------------------
-## Initiate Pubnub State
-## -----------------------------------------------------------------------
-pubnub = Pubnub( publish_key=publish_key, subscribe_key=subscribe_key,
- secret_key=secret_key, cipher_key=cipher_key, ssl_on=ssl_on)
-crazy = 'hello_world'
-
-
-print pubnub.here_now( {
- 'channel' : crazy
-})
-
diff --git a/python/3.3/history-example.py b/python/3.3/history-example.py
deleted file mode 100755
index cedf69e..0000000
--- a/python/3.3/history-example.py
+++ /dev/null
@@ -1,12 +0,0 @@
-from Pubnub import Pubnub
-
-## Initiat Class
-pubnub = Pubnub( 'demo', 'demo', None, False )
-
-## History Example
-history = pubnub.history({
- 'channel' : 'hello_world',
- 'limit' : 1
-})
-print(history)
-
diff --git a/python/3.3/publish-example.py b/python/3.3/publish-example.py
deleted file mode 100755
index c97034b..0000000
--- a/python/3.3/publish-example.py
+++ /dev/null
@@ -1,14 +0,0 @@
-from Pubnub import Pubnub
-
-## Initiate Class
-pubnub = Pubnub( publish_key='demo', subscribe_key='demo', ssl_on=False )
-
-## Publish Example
-info = pubnub.publish({
- 'channel' : 'hello_world',
- 'message' : {
- 'some_text' : 'Hello my World'
- }
-})
-print(info)
-
diff --git a/python/3.3/subscribe-example.py b/python/3.3/subscribe-example.py
deleted file mode 100755
index e458e2b..0000000
--- a/python/3.3/subscribe-example.py
+++ /dev/null
@@ -1,64 +0,0 @@
-import sys
-import threading
-import time
-import random
-import string
-from Pubnub import Pubnub
-
-## Initiate Class
-pubnub = Pubnub( 'demo', 'demo', None, False )
-
-print("My UUID is: "+pubnub.uuid)
-
-channel = ''.join(random.choice(string.ascii_letters + string.digits) for x in range(20))
-
-## Subscribe Example
-def receive(message) :
- print(message)
- return False
-
-def pres_event(message):
- print(message)
- return False
-
-def subscribe():
- print("Listening for messages on '%s' channel..." % channel)
- pubnub.subscribe({
- 'channel' : channel,
- 'callback' : receive
- })
-
-def presence():
- print("Listening for presence events on '%s' channel..." % channel)
- pubnub.presence({
- 'channel' : channel,
- 'callback' : pres_event
- })
-
-def publish():
- print("Publishing a test message on '%s' channel..." % channel)
- pubnub.publish({
- 'channel' : channel,
- 'message' : { 'text':'foo bar' }
- })
-
-pres_thread = threading.Thread(target=presence)
-pres_thread.daemon=True
-pres_thread.start()
-
-sub_thread = threading.Thread(target=subscribe)
-sub_thread.daemon=True
-sub_thread.start()
-
-time.sleep(3)
-
-publish()
-
-
-print("waiting for subscribes and presence")
-pres_thread.join()
-
-print pubnub.here_now({'channel':channel})
-
-sub_thread.join()
-
diff --git a/python/3.3/unit-test.py b/python/3.3/unit-test.py
deleted file mode 100755
index 88391a0..0000000
--- a/python/3.3/unit-test.py
+++ /dev/null
@@ -1,77 +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.0 Real-time Push Cloud API
-## -----------------------------------
-
-from Pubnub import Pubnub
-import sys
-
-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
-ssl_on = len(sys.argv) > 4 and bool(sys.argv[4]) or False
-
-
-## -----------------------------------------------------------------------
-## Initiat Class
-## -----------------------------------------------------------------------
-
-pubnub = Pubnub( publish_key, subscribe_key, secret_key, ssl_on )
-crazy = ' ~`!@#$%^&*(顶顅Ȓ)+=[]\\{}|;\':",./<>?abcd'
-
-## ---------------------------------------------------------------------------
-## Unit Test Function
-## ---------------------------------------------------------------------------
-def test( trial, name ) :
- if trial :
- print( 'PASS: ' + name )
- else :
- print( 'FAIL: ' + name )
-
-## -----------------------------------------------------------------------
-## Publish Example
-## -----------------------------------------------------------------------
-pubish_success = pubnub.publish({
- 'channel' : crazy,
- 'message' : crazy
-})
-test( pubish_success[0] == 1, 'Publish First Message Success' )
-
-## -----------------------------------------------------------------------
-## History Example
-## -----------------------------------------------------------------------
-history = pubnub.history({
- 'channel' : crazy,
- 'limit' : 1
-})
-test(
- history[0].encode('utf-8') == crazy,
- 'History Message: ' + history[0]
-)
-test( len(history) == 1, 'History Message Count' )
-
-## -----------------------------------------------------------------------
-## PubNub Server Time Example
-## -----------------------------------------------------------------------
-timestamp = pubnub.time()
-test( timestamp > 0, 'PubNub Server Time: ' + str(timestamp) )
-
-## -----------------------------------------------------------------------
-## Subscribe Example
-## -----------------------------------------------------------------------
-def receive(message) :
- print(message)
- return True
-
-pubnub.subscribe({
- 'channel' : crazy,
- 'callback' : receive
-})
-
-
generated by cgit v1.2.3 (git 2.25.1) at 2025-10-31 17:10:43 +0000