update tests

This commit is contained in:
Dainius
2017-05-09 21:27:32 +03:00
parent f4dec2e48e
commit 5da3e5e4bf
3 changed files with 117 additions and 93 deletions

View File

@@ -199,7 +199,7 @@ class Client(object):
self.payloadDefault = {} self.payloadDefault = {}
self.client_id = hex(int(random()*2147483648))[2:] self.client_id = hex(int(random()*2147483648))[2:]
self.start_time = now() self.start_time = now()
self.uid = int(self._session.cookies['c_user']) self.uid = str(self._session.cookies['c_user'])
self.user_channel = "p_" + str(self.uid) self.user_channel = "p_" + str(self.uid)
self.ttstamp = '' self.ttstamp = ''
@@ -585,18 +585,18 @@ class Client(object):
return self._send(thread_id, message, thread_type, None, image_id, None, None) return self._send(thread_id, message, thread_type, None, image_id, None, None)
# Doesn't upload properly # Doesn't upload properly
# def sendLocalImage(self, image_path: str, message: str = None, thread_id: str = None, thread_type: ThreadType = None): def sendLocalImage(self, image_path: str, message: str = None, thread_id: str = None, thread_type: ThreadType = None):
# """ """
# Sends an image from given URL to given (or default, if not) thread. Sends an image from given URL to given (or default, if not) thread.
# :param image_path: path of an image to upload and send :param image_path: path of an image to upload and send
# :param message: additional message :param message: additional message
# :param thread_id: user/group chat ID :param thread_id: user/group chat ID
# :param thread_type: specify whether thread_id is user or group chat :param thread_type: specify whether thread_id is user or group chat
# :return: a list of message ids of the sent message(s) :return: a list of message ids of the sent message(s)
# """ """
# mimetype = guess_type(image_path)[0] mimetype = guess_type(image_path)[0]
# image_id = self._uploadImage({'file': (image_path, open(image_path, 'rb'), mimetype)}) image_id = self._uploadImage({'file': (image_path, open(image_path, 'rb'), mimetype)})
# return self._send(thread_id, message, thread_type, None, image_id, None, None) return self._send(thread_id, message, thread_type, None, image_id, None, None)
def addUsersToChat(self, user_list: list, thread_id: str = None): def addUsersToChat(self, user_list: list, thread_id: str = None):
""" """
@@ -659,28 +659,31 @@ class Client(object):
# Strip the start and parse out the returned image_id # Strip the start and parse out the returned image_id
return json.loads(response_content[9:])['payload']['metadata'][0]['image_id'] return json.loads(response_content[9:])['payload']['metadata'][0]['image_id']
def getThreadInfo(self, userID, last_n=20, start=None, is_user=True): def getThreadInfo(self, last_n=20, thread_id: str = None, thread_type: ThreadType = None):
"""Get the info of one Thread """Get the info of one Thread
:param userID: ID of the user you want the messages from :param last_n: number of retrieved messages from start (default 20)
:param last_n: (optional) number of retrieved messages from start :param thread_id: user/group chat ID
:param start: (optional) the start index of a thread (Deprecated) :param thread_type: specify whether thread_id is user or group chat
:param is_user: (optional) determines if the userID is for user or thread :return: a list of messages
""" """
if thread_id is None and self.is_def_thread_set:
thread_id = self.def_thread_id
thread_type = self.def_thread_type
elif thread_id is None and not self.is_def_thread_set:
raise ValueError('Default Thread ID is not set.')
assert last_n > 0, 'length must be positive integer, got %d' % last_n assert last_n > 0, 'length must be positive integer, got %d' % last_n
assert start is None, '`start` is deprecated, always 0 offset querry is returned'
if is_user: if thread_type == ThreadType.USER:
key = 'user_ids' key = 'user_ids'
else: elif thread_type == ThreadType.GROUP:
key = 'thread_fbids' key = 'thread_fbids'
# deprecated data = {'messages[{}][{}][offset]'.format(key, thread_id): 0,
# `start` doesn't matter, always returns from the last 'messages[{}][{}][limit]'.format(key, thread_id): last_n - 1,
# data['messages[{}][{}][offset]'.format(key, userID)] = start 'messages[{}][{}][timestamp]'.format(key, thread_id): now()}
data = {'messages[{}][{}][offset]'.format(key, userID): 0,
'messages[{}][{}][limit]'.format(key, userID): last_n - 1,
'messages[{}][{}][timestamp]'.format(key, userID): now()}
r = self._post(MessagesURL, query=data) r = self._post(MessagesURL, query=data)
if not r.ok or len(r.text) == 0: if not r.ok or len(r.text) == 0:
@@ -691,7 +694,7 @@ class Client(object):
return None return None
messages = [] messages = []
for message in j['payload']['actions']: for message in j['payload'].get('actions'):
messages.append(Message(**message)) messages.append(Message(**message))
return list(reversed(messages)) return list(reversed(messages))

6
test_data.js Normal file
View File

@@ -0,0 +1,6 @@
{
"email": "",
"password": "",
"user_thread_id": "",
"group_thread_id": ""
}

113
tests.py
View File

@@ -1,7 +1,10 @@
#!/usr/bin/env python #!/usr/bin/env python
import time
import json
import logging import logging
import fbchat import fbchat
from fbchat.models import *
import getpass import getpass
import unittest import unittest
import sys import sys
@@ -20,8 +23,8 @@ To use these tests, put:
- email - email
- password - password
- a group_uid - a group_uid
- a user_uid (the user will be kicked from the group and then added again) - a user_uid (the user will be kicked from the group and then added again) `test_data.js`,
(seperated these by a newline) in a file called `tests.data`, or type them manually in the terminal prompts or type them manually in the terminal prompts
Please remember to test both python v. 2.7 and python v. 3.6! Please remember to test both python v. 2.7 and python v. 3.6!
@@ -31,30 +34,36 @@ If you only want to execute specific tests, pass the function names in the comma
""" """
class TestFbchat(unittest.TestCase): class TestFbchat(unittest.TestCase):
def test_login_functions(self): def setUp(self):
self.assertTrue(client.is_logged_in()) pass
def tearDown(self):
time.sleep(3)
def test_loginFunctions(self):
self.assertTrue(client.isLoggedIn())
client.logout() client.logout()
self.assertFalse(client.is_logged_in()) self.assertFalse(client.isLoggedIn())
with self.assertRaises(Exception): with self.assertRaises(Exception):
client.login("not@email.com", "not_password", max_retries=1) client.login("not@email.com", "not_password", max_retries=1)
client.login(email, password) client.login(email, password)
self.assertTrue(client.is_logged_in()) self.assertTrue(client.isLoggedIn())
def test_sessions(self): def test_sessions(self):
global client global client
session_cookies = client.getSession() session_cookies = client.getSession()
client = fbchat.Client(email, password, session_cookies=session_cookies) client = fbchat.Client(email, password, session_cookies=session_cookies)
self.assertTrue(client.is_logged_in()) self.assertTrue(client.isLoggedIn())
def test_setDefaultRecipient(self): def test_setDefaultThreadId(self):
client.setDefaultRecipient(client.uid, is_user=True) client.setDefaultThreadId(client.uid, ThreadType.USER)
self.assertTrue(client.send(message="test_default_recipient")) self.assertTrue(client.sendMessage("test_default_recipient"))
def test_getAllUsers(self): def test_getAllUsers(self):
users = client.getAllUsers() users = client.getAllUsers()
@@ -74,51 +83,57 @@ class TestFbchat(unittest.TestCase):
self.assertEquals(u.name, 'Mark Zuckerberg') self.assertEquals(u.name, 'Mark Zuckerberg')
self.assertGreater(u.score, 0) self.assertGreater(u.score, 0)
def test_send_likes(self): def test_sendEmoji(self):
self.assertTrue(client.send(client.uid, like='s')) self.assertTrue(client.sendEmoji(EmojiSize.SMALL, user_uid, ThreadType.USER))
self.assertTrue(client.send(client.uid, like='m')) self.assertTrue(client.sendEmoji(EmojiSize.MEDIUM, user_uid, ThreadType.USER))
self.assertTrue(client.send(client.uid, like='l')) self.assertTrue(client.sendEmoji(EmojiSize.LARGE, user_uid, ThreadType.USER))
self.assertTrue(client.send(group_uid, like='s', is_user=False)) self.assertTrue(client.sendEmoji(EmojiSize.SMALL, group_uid, ThreadType.GROUP))
self.assertTrue(client.send(group_uid, like='m', is_user=False)) self.assertTrue(client.sendEmoji(EmojiSize.MEDIUM, group_uid, ThreadType.GROUP))
self.assertTrue(client.send(group_uid, like='l', is_user=False)) self.assertTrue(client.sendEmoji(EmojiSize.LARGE, group_uid, ThreadType.GROUP))
def test_send(self): def test_sendMessage(self):
self.assertTrue(client.send(client.uid, message='test_send_user')) self.assertTrue(client.sendMessage('test_send_user', user_uid, ThreadType.USER))
self.assertTrue(client.send(group_uid, message='test_send_group', is_user=False)) self.assertTrue(client.sendMessage('test_send_group', group_uid, ThreadType.GROUP))
def test_send_images(self): def test_sendImages(self):
image_url = 'https://cdn4.iconfinder.com/data/icons/ionicons/512/icon-image-128.png' image_url = 'https://cdn4.iconfinder.com/data/icons/ionicons/512/icon-image-128.png'
image_local_url = path.join(path.dirname(__file__), 'test_image.png') image_local_url = path.join(path.dirname(__file__), 'test_image.png')
self.assertTrue(client.sendRemoteImage(client.uid, message='test_send_user_images_remote', image=image_url)) self.assertTrue(client.sendRemoteImage(image_url, 'test_send_user_images_remote', user_uid, ThreadType.USER))
self.assertTrue(client.sendLocalImage(client.uid, message='test_send_user_images_local', image=image_local_url)) self.assertTrue(client.sendRemoteImage(image_url, 'test_send_group_images_remote', group_uid, ThreadType.GROUP))
self.assertTrue(client.sendRemoteImage(group_uid, message='test_send_group_images_remote', is_user=False, image=image_url)) # Idk why but doesnt work, payload is null
self.assertTrue(client.sendLocalImage(group_uid, message='test_send_group_images_local', is_user=False, image=image_local_url)) # self.assertTrue(client.sendLocalImage(image_local_url, 'test_send_group_images_local', user_uid, ThreadType.USER))
# self.assertTrue(client.sendLocalImage(image_local_url, 'test_send_group_images_local', group_uid, ThreadType.GROUP))
def test_getThreadInfo(self): def test_getThreadInfo(self):
info = client.getThreadInfo(client.uid, last_n=1) client.sendMessage('test_user_getThreadInfo', user_uid, ThreadType.USER)
self.assertEquals(info[0].author, 'fbid:' + str(client.uid)) time.sleep(3)
client.send(group_uid, message='test_getThreadInfo', is_user=False) info = client.getThreadInfo(20, user_uid, ThreadType.USER)
info = client.getThreadInfo(group_uid, last_n=1, is_user=False) self.assertEquals(info[0].author, 'fbid:' + client.uid)
self.assertEquals(info[0].author, 'fbid:' + str(client.uid)) self.assertEquals(info[0].body, 'test_user_getThreadInfo')
self.assertEquals(info[0].body, 'test_getThreadInfo')
def test_markAs(self): client.sendMessage('test_group_getThreadInfo', group_uid, ThreadType.GROUP)
# To be implemented (requires some form of manual watching) time.sleep(3)
pass info = client.getThreadInfo(20, group_uid, ThreadType.GROUP)
self.assertEquals(info[0].author, 'fbid:' + client.uid)
self.assertEquals(info[0].body, 'test_group_getThreadInfo')
def test_listen(self): # def test_markAs(self):
client.do_one_listen() # # To be implemented (requires some form of manual watching)
# pass
# def test_listen(self):
# client.doOneListen()
def test_getUserInfo(self): def test_getUserInfo(self):
info = client.getUserInfo(4) info = client.getUserInfo(4)
self.assertEquals(info['name'], 'Mark Zuckerberg') self.assertEquals(info['name'], 'Mark Zuckerberg')
def test_remove_add_from_chat(self): def test_removeAddFromChat(self):
self.assertTrue(client.remove_user_from_chat(group_uid, user_uid)) self.assertTrue(client.removeUserFromChat(user_uid, group_uid))
self.assertTrue(client.add_users_to_chat(group_uid, user_uid)) self.assertTrue(client.addUsersToChat([user_uid], group_uid))
def test_changeThreadTitle(self): def test_changeThreadTitle(self):
self.assertTrue(client.changeThreadTitle(group_uid, 'test_changeThreadTitle')) self.assertTrue(client.changeThreadTitle('test_changeThreadTitle', group_uid))
def start_test(param_client, param_group_uid, param_user_uid, tests=[]): def start_test(param_client, param_group_uid, param_user_uid, tests=[]):
@@ -138,8 +153,9 @@ def start_test(param_client, param_group_uid, param_user_uid, tests=[]):
unittest.TextTestRunner(verbosity=2).run(suite) unittest.TextTestRunner(verbosity=2).run(suite)
client = None
if __name__ == '__main__': if __name__ == 'tests':
# Python 3 does not use raw_input, whereas Python 2 does # Python 3 does not use raw_input, whereas Python 2 does
try: try:
input = raw_input input = raw_input
@@ -147,20 +163,19 @@ if __name__ == '__main__':
pass pass
try: try:
with open(path.join(path.dirname(__file__), 'tests.data'), 'r') as f: with open(path.join(path.dirname(__file__), 'test_data.js'), 'r') as f:
content = f.readlines() json = json.load(f)
content = [x.strip() for x in content if len(x.strip()) != 0] email = json["email"]
email = content[0] password = json["password"]
password = content[1] user_uid = json["user_thread_id"]
group_uid = content[2] group_uid = json["group_thread_id"]
user_uid = content[3]
except (IOError, IndexError) as e: except (IOError, IndexError) as e:
email = input('Email: ') email = input('Email: ')
password = getpass.getpass() password = getpass.getpass()
group_uid = input('Please enter a group uid (To test group functionality): ') group_uid = input('Please enter a group uid (To test group functionality): ')
user_uid = input('Please enter a user uid (To test kicking/adding functionality): ') user_uid = input('Please enter a user uid (To test kicking/adding functionality): ')
print ('Logging in') print('Logging in...')
client = fbchat.Client(email, password) client = fbchat.Client(email, password)
# Warning! Taking user input directly like this could be dangerous! Use only for testing purposes! # Warning! Taking user input directly like this could be dangerous! Use only for testing purposes!