class Client:
"""This is the client class to connect to Diaspora.
-
"""
def __init__(self, pod, username, password):
"""
:returns: list -- list of Post objects.
"""
- data = {'authenticity_token': self.get_token()}
r = self._sessionget('stream.json')
if r.status_code != 200:
:returns: list -- list of json formatted notifications
"""
- data = {'authenticity_token': self.get_token()}
r = self._sessionget('notifications.json')
if r.status_code != 200:
:returns: list -- list of Post objects
"""
- data = {'authenticity_token': self.get_token()}
r = self._sessionget('mentions.json')
if r.status_code != 200:
:returns: list -- list of Post objects
"""
- data = {'authenticity_token': self.get_token()}
r = self._sessionget('tags/{0}.json'.format(tag))
if r.status_code != 200:
tagged_posts = r.json()
return [ diaspy.models.Post(str(post['id']), self) for post in tagged_posts ]
+ def get_mailbox(self):
+ """This functions returns a list of messages found in the conversation.
+
+ :returns: list -- list of Conversation objects.
+ """
+ r = self._sessionget('conversations.json')
+
+ if r.status_code != 200:
+ raise Exception('wrong status code: {0}'.format(r.status_code))
+
+ mailbox = r.json()
+ return [ diaspy.conversations.Conversation(str(conversation['conversation']['id']), self) for conversation in mailbox ]
+
def add_user_to_aspect(self, user_id, aspect_id):
""" this function adds a user to an aspect.
def remove_aspect(self, aspect_id):
""" This function adds a new aspect.
"""
-
data = {'authenticity_token': self.get_token()}
r = self.session.delete('{0}/aspects/{1}'.format(self.pod, aspect_id),
if r.status_code != 404:
raise Exception('wrong status code: {0}'.format(r.status_code))
- def get_mailbox(self):
- """This functions returns a list of messages found in the conversation.
-
- :returns: list -- list of Conversation objects.
- """
- data = {'authenticity_token': self.get_token()}
- r = self._sessionget('conversations.json')
-
- if r.status_code != 200:
- raise Exception('wrong status code: {0}'.format(r.status_code))
-
- mailbox = r.json()
- return [ diaspy.conversations.Conversation(str(conversation['conversation']['id']), self) for conversation in mailbox ]
-
def new_conversation(self, contacts, subject, text):
- """ start a new conversation
+ """Start a new conversation.
:param contacts: recipients ids, no guids, comma sperated.
:type contacts: str
:type subject: str
:param text: text of the message.
:type text: str
-
"""
-
data = {'contact_ids': contacts,
'conversation[subject]': subject,
'conversation[text]': text,
self.assertEqual(client._login_data['user[password]'], __passwd__)
self.assertEqual(client._login_data['authenticity_token'], client.get_token())
+ def testGettingUserInfo(self):
+ client = diaspy.client.Client(__pod__, __username__, __passwd__)
+ info = client.get_user_info()
+ self.assertEqual(dict, type(info))
+
+ def testGettingStream(self):
+ client = diaspy.client.Client(__pod__, __username__, __passwd__)
+ stream = client.get_stream()
+ self.assertEqual(list, type(stream))
+ if stream: self.assertEqual(diaspy.models.Post, type(stream[0]))
+
+ def testGettingNotifications(self):
+ client = diaspy.client.Client(__pod__, __username__, __passwd__)
+ notifications = client.get_notifications()
+ self.assertEqual(list, type(notifications))
+ if notifications: self.assertEqual(dict, type(notifications[0]))
+
def testGettingTag(self):
client = diaspy.client.Client(pod=__pod__, username=__username__, password=__passwd__)
tag = client.get_tag('foo')