From 151a1142355eb1fac126df79529c7d6711c55332 Mon Sep 17 00:00:00 2001 From: Mads Marquart Date: Mon, 4 Jun 2018 13:44:04 +0200 Subject: [PATCH] TravisCI integration and updated test suite (#296) * Make TravisCI setup * Use pytest, move tests to seperate files * Added system to check if `onX` events were successfully executed --- .gitignore | 1 + .travis.yml | 35 ++++ pytest.ini | 6 + requirements.txt | 1 + tests.py | 261 ------------------------------ tests/conftest.py | 101 ++++++++++++ tests/test_base.py | 55 +++++++ tests/test_fetch.py | 79 +++++++++ tests/test_message_management.py | 12 ++ tests/test_search.py | 18 +++ tests/test_send.py | 110 +++++++++++++ tests/test_tests.py | 12 ++ tests/test_thread_interraction.py | 112 +++++++++++++ tests/utils.py | 83 ++++++++++ 14 files changed, 625 insertions(+), 261 deletions(-) create mode 100644 .travis.yml create mode 100644 pytest.ini delete mode 100644 tests.py create mode 100644 tests/conftest.py create mode 100644 tests/test_base.py create mode 100644 tests/test_fetch.py create mode 100644 tests/test_message_management.py create mode 100644 tests/test_search.py create mode 100644 tests/test_send.py create mode 100644 tests/test_tests.py create mode 100644 tests/test_thread_interraction.py create mode 100644 tests/utils.py diff --git a/.gitignore b/.gitignore index 2345f05..dd3713c 100644 --- a/.gitignore +++ b/.gitignore @@ -29,6 +29,7 @@ my_tests.py my_test_data.json my_data.json tests.data +.pytest_cache # Virtual environment venv/ diff --git a/.travis.yml b/.travis.yml new file mode 100644 index 0000000..1b354d3 --- /dev/null +++ b/.travis.yml @@ -0,0 +1,35 @@ +sudo: false +language: python +python: +- 2.7 +- 3.4 +- 3.5 +- 3.6 + +env: + global: + # These two accounts are made specifically for this purpose, and the passwords are really only encrypted for obscurity + # In reality, you could probably still login with the accounts by looking at log output from travis-ci.org + - client1_email=travis.fbchat1@gmail.com # Facebook ID: 100023782141139 + - secure: "JrQ8mrE7S8ActFiQF5ulICi9jXrluYTuEM0cHphI1N/Zj4pEw+k24hVhfGJvLLuIDRV12DxScc7dKgUIvqavc2Ko9cy0Ge7GTkjAjGYiyCPabxUUjgtPB173pK0x9nkjmyXdTbb2P4ZsaIfxPwSGPWS77jEkEboqia8Cqr/RhVA=" # client1_password + - client2_email=fbchat.travis2@gmail.com # Facebook ID: 100026538491708 + - secure: "AGQfwIm6VP3EosnP/Mu7Em0+gb5638+INMqXaJ8HEr2kkV0c12C2G2aDRzsD7DjinKTfGsPAMvY3HPmtx8luU7urQeM55ArmOgeZbbLK+I+hmm08lubBc/Ik3OSehZc0u7IqKIHTxOxlYMzCgZCFQYT0QtOVZUt8Q529AzbUVv4=" # client2_password + - group_id=1463789480385605 + +before_script: + - if [[ "$TRAVIS_PYTHON_VERSION" = "2.7" ]]; then export PYTEST_ADDOPTS='-m ""'; fi; # expensive tests (otherwise disabled in pytest.ini) + - if [[ "$TRAVIS_PULL_REQUEST" != false ]]; then export PYTEST_ADDOPTS='-m offline'; fi; # offline tests only + +script: python -m pytest || python -m pytest --lf; # Run failed tests twice + +cache: pip + +deploy: + provider: pypi + user: madsmtm + password: + secure: "fH7+JhSnhIh1FZPgt+TE5J+REVCHaasEuhXZmpV8NgOrIZMbTtGOAfbWcwZWNrgoODN+adZFAS0OMDBEKL4WDudM4aXtrX3W4efy4Btmrn10YiAIgGQhG2HtQeM8HMAWX6wyY+ZnQz2NcJYL/SQqRWfRnHtMJlQVYErzqubdB7Y=" + on: + python: 3.6 + branch: master + tags: true diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..ecbb1b3 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,6 @@ +[pytest] +xfail_strict=true +markers = + offline: Offline tests, aka. tests that can be executed without the need of a client + expensive: Expensive tests, which should be executed sparingly +addopts = -m "not expensive" diff --git a/requirements.txt b/requirements.txt index 22069fb..5e641fc 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ requests lxml beautifulsoup4 enum34; python_version < '3.4' +six diff --git a/tests.py b/tests.py deleted file mode 100644 index caf6748..0000000 --- a/tests.py +++ /dev/null @@ -1,261 +0,0 @@ -#!/usr/bin/env python -# -*- coding: UTF-8 -*- - -from __future__ import unicode_literals -import json -import logging -import unittest -from getpass import getpass -from sys import argv -from os import path, chdir -from glob import glob -from fbchat import Client -from fbchat.models import * -import py_compile - -logging_level = logging.ERROR - -""" - -Testing script for `fbchat`. -Full documentation on https://fbchat.readthedocs.io/ - -""" - -test_sticker_id = '767334476626295' - -class CustomClient(Client): - def __init__(self, *args, **kwargs): - self.got_qprimer = False - super(type(self), self).__init__(*args, **kwargs) - - def onQprimer(self, msg, **kwargs): - self.got_qprimer = True - -class TestFbchat(unittest.TestCase): - def test_examples(self): - # Checks for syntax errors in the examples - chdir('examples') - for f in glob('*.txt'): - print(f) - with self.assertRaises(py_compile.PyCompileError): - py_compile.compile(f) - - chdir('..') - - def test_loginFunctions(self): - self.assertTrue(client.isLoggedIn()) - - client.logout() - - self.assertFalse(client.isLoggedIn()) - - with self.assertRaises(Exception): - client.login('', '', max_tries=1) - - client.login(email, password) - - self.assertTrue(client.isLoggedIn()) - - def test_sessions(self): - global client - session_cookies = client.getSession() - client = CustomClient(email, password, session_cookies=session_cookies, logging_level=logging_level) - - self.assertTrue(client.isLoggedIn()) - - def test_defaultThread(self): - # setDefaultThread - for thread in threads: - client.setDefaultThread(thread_id=thread['id'], thread_type=thread['type']) - self.assertTrue(client.send(Message(text='test_default_recipient★'))) - - # resetDefaultThread - client.resetDefaultThread() - with self.assertRaises(ValueError): - client.send(Message(text='should_not_send')) - - def test_fetchAllUsers(self): - users = client.fetchAllUsers() - self.assertGreater(len(users), 0) - - def test_searchFor(self): - users = client.searchForUsers('Mark Zuckerberg') - self.assertGreater(len(users), 0) - - u = users[0] - - # Test if values are set correctly - self.assertEqual(u.uid, '4') - self.assertEqual(u.type, ThreadType.USER) - self.assertEqual(u.photo[:4], 'http') - self.assertEqual(u.url[:4], 'http') - self.assertEqual(u.name, 'Mark Zuckerberg') - - group_name = client.changeThreadTitle('tést_searchFor', thread_id=group_id, thread_type=ThreadType.GROUP) - groups = client.searchForGroups('té') - self.assertGreater(len(groups), 0) - - def test_send(self): - for thread in threads: - client.setDefaultThread(thread_id=thread['id'], thread_type=thread['type']) - - self.assertIsNotNone(client.send(Message(emoji_size=EmojiSize.SMALL))) - self.assertIsNotNone(client.send(Message(emoji_size=EmojiSize.MEDIUM))) - self.assertIsNotNone(client.send(Message(text='😆', emoji_size=EmojiSize.LARGE))) - - self.assertIsNotNone(client.send(Message(text='test_send★'))) - with self.assertRaises(FBchatFacebookError): - self.assertIsNotNone(client.send(Message(text='test_send_should_fail★'), thread_id=thread['id'], thread_type=(ThreadType.GROUP if thread['type'] == ThreadType.USER else ThreadType.USER))) - - self.assertIsNotNone(client.send(Message(text='Hi there @user', mentions=[Mention(user_id, offset=9, length=5)]))) - self.assertIsNotNone(client.send(Message(text='Hi there @group', mentions=[Mention(group_id, offset=9, length=6)]))) - - self.assertIsNotNone(client.send(Message(sticker=Sticker(test_sticker_id)))) - - def test_sendImages(self): - image_url = 'https://github.com/carpedm20/fbchat/raw/master/tests/image.png' - image_local_url = path.join(path.dirname(__file__), 'tests/image.png') - for thread in threads: - client.setDefaultThread(thread_id=thread['id'], thread_type=thread['type']) - mentions = [Mention(thread['id'], offset=26, length=4)] - self.assertTrue(client.sendRemoteImage(image_url, Message(text='test_send_image_remote_to_@you★', mentions=mentions))) - self.assertTrue(client.sendLocalImage(image_local_url, Message(text='test_send_image_local_to__@you★', mentions=mentions))) - - def test_fetchThreadList(self): - threads = client.fetchThreadList(limit=2) - self.assertEqual(len(threads), 2) - - def test_fetchThreadMessages(self): - for thread in threads: - client.setDefaultThread(thread_id=thread['id'], thread_type=thread['type']) - client.send(Message(text='test_getThreadInfo★')) - - messages = client.fetchThreadMessages(limit=1) - self.assertEqual(messages[0].author, client.uid) - self.assertEqual(messages[0].text, 'test_getThreadInfo★') - - def test_listen(self): - client.startListening() - client.doOneListen() - client.stopListening() - - self.assertTrue(client.got_qprimer) - - def test_fetchInfo(self): - info = client.fetchUserInfo('4')['4'] - self.assertEqual(info.name, 'Mark Zuckerberg') - - info = client.fetchGroupInfo(group_id)[group_id] - self.assertEqual(info.type, ThreadType.GROUP) - - def test_removeAddFromGroup(self): - client.removeUserFromGroup(user_id, thread_id=group_id) - client.addUsersToGroup(user_id, thread_id=group_id) - - def test_changeThreadTitle(self): - for thread in threads: - client.changeThreadTitle('test_changeThreadTitle★', thread_id=thread['id'], thread_type=thread['type']) - - def test_changeNickname(self): - for thread in threads: - client.setDefaultThread(thread_id=thread['id'], thread_type=thread['type']) - client.changeNickname('test_changeNicknameSelf★', client.uid) - client.changeNickname('test_changeNicknameOther★', user_id) - - def test_changeThreadEmoji(self): - for thread in threads: - client.changeThreadEmoji('😀', thread_id=thread['id']) - client.changeThreadEmoji('😀', thread_id=thread['id']) - - def test_changeThreadColor(self): - for thread in threads: - client.setDefaultThread(thread_id=thread['id'], thread_type=thread['type']) - client.changeThreadColor(ThreadColor.BRILLIANT_ROSE) - client.changeThreadColor(ThreadColor.MESSENGER_BLUE) - - def test_reactToMessage(self): - for thread in threads: - mid = client.send(Message(text='test_reactToMessage★'), thread_id=thread['id'], thread_type=thread['type']) - client.reactToMessage(mid, MessageReaction.LOVE) - - def test_setTypingStatus(self): - for thread in threads: - client.setDefaultThread(thread_id=thread['id'], thread_type=thread['type']) - client.setTypingStatus(TypingStatus.TYPING) - client.setTypingStatus(TypingStatus.STOPPED) - - -def start_test(param_client, param_group_id, param_user_id, param_threads, tests=[]): - global client - global group_id - global user_id - global threads - - client = param_client - group_id = param_group_id - user_id = param_user_id - threads = param_threads - - tests = ['test_' + test if 'test_' != test[:5] else test for test in tests] - - if len(tests) == 0: - suite = unittest.TestLoader().loadTestsFromTestCase(TestFbchat) - else: - suite = unittest.TestSuite(map(TestFbchat, tests)) - print('Starting test(s)') - unittest.TextTestRunner(verbosity=2).run(suite) - - -client = None - -if __name__ == '__main__': - # Python 3 does not use raw_input, whereas Python 2 does - try: - input = raw_input - except Exception as e: - pass - - try: - with open(path.join(path.dirname(__file__), 'tests/my_data.json'), 'r') as f: - j = json.load(f) - email = j['email'] - password = j['password'] - user_id = j['user_thread_id'] - group_id = j['group_thread_id'] - session = j.get('session') - except (IOError, IndexError) as e: - email = input('Email: ') - password = getpass() - group_id = input('Please enter a group thread id (To test group functionality): ') - user_id = input('Please enter a user thread id (To test kicking/adding functionality): ') - threads = [ - { - 'id': user_id, - 'type': ThreadType.USER - }, - { - 'id': group_id, - 'type': ThreadType.GROUP - } - ] - - print('Logging in...') - client = CustomClient(email, password, logging_level=logging_level, session_cookies=session) - - # Warning! Taking user input directly like this could be dangerous! Use only for testing purposes! - start_test(client, group_id, user_id, threads, argv[1:]) - - with open(path.join(path.dirname(__file__), 'tests/my_data.json'), 'w') as f: - session = None - try: - session = client.getSession() - except Exception: - print('Unable to fetch client session!') - json.dump({ - 'email': email, - 'password': password, - 'user_thread_id': user_id, - 'group_thread_id': group_id, - 'session': session - }, f) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..879e3dc --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,101 @@ +# -*- coding: utf-8 -*- + +from __future__ import unicode_literals + +import pytest +import json + +from utils import * +from contextlib import contextmanager +from fbchat.models import ThreadType + + +@pytest.fixture(scope="session") +def user(client2): + return {"id": client2.uid, "type": ThreadType.USER} + + +@pytest.fixture(scope="session") +def group(pytestconfig): + return {"id": load_variable("group_id", pytestconfig.cache), "type": ThreadType.GROUP} + + +@pytest.fixture(scope="session", params=["user", "group"]) +def thread(request, user, group): + return user if request.param == "user" else group + + +@pytest.fixture(scope="session") +def client1(pytestconfig): + with load_client(1, pytestconfig.cache) as c: + yield c + + +@pytest.fixture(scope="session") +def client2(pytestconfig): + with load_client(2, pytestconfig.cache) as c: + yield c + + +@pytest.fixture # (scope="session") +def client(client1, thread): + client1.setDefaultThread(thread["id"], thread["type"]) + yield client1 + client1.resetDefaultThread() + + +@pytest.fixture(scope="session", params=["client1", "client2"]) +def client_all(request, client1, client2): + return client1 if request.param == "client1" else client2 + + +@pytest.fixture(scope="session") +def catch_event(client2): + t = ClientThread(client2) + t.start() + + @contextmanager + def inner(method_name): + caught = CaughtValue() + old_method = getattr(client2, method_name) + + # Will be called by the other thread + def catch_value(*args, **kwargs): + old_method(*args, **kwargs) + # Make sure the `set` is only called once + if not caught.is_set(): + caught.set(kwargs) + + setattr(client2, method_name, catch_value) + yield caught + caught.wait() + if not caught.is_set(): + raise ValueError("The value could not be caught") + setattr(client2, method_name, old_method) + + yield inner + + t.should_stop.set() + + try: + # Make the client send a messages to itself, so the blocking pull request will return + # This is probably not safe, since the client is making two requests simultaneously + client2.sendMessage("Shutdown", client2.uid) + finally: + t.join() + + +@pytest.fixture # (scope="session") +def compare(client, thread): + def inner(caught_event, **kwargs): + d = { + "author_id": client.uid, + "thread_id": client.uid + if thread["type"] == ThreadType.USER + else thread["id"], + "thread_type": thread["type"], + } + d.update(kwargs) + return subset(caught_event.res, **d) + + return inner diff --git a/tests/test_base.py b/tests/test_base.py new file mode 100644 index 0000000..6a60482 --- /dev/null +++ b/tests/test_base.py @@ -0,0 +1,55 @@ +# -*- coding: utf-8 -*- + +from __future__ import unicode_literals + +import pytest +import py_compile + +from glob import glob +from os import path, environ +from fbchat import Client +from fbchat.models import FBchatUserError, Message + + +@pytest.mark.offline +def test_examples(): + # Compiles the examples, to check for syntax errors + for name in glob(path.join(path.dirname(__file__), "../examples", "*.py")): + py_compile.compile(name) + + +@pytest.mark.trylast +@pytest.mark.expensive +def test_login(client1): + assert client1.isLoggedIn() + email = client1.email + password = client1.password + + client1.logout() + + assert not client1.isLoggedIn() + + with pytest.raises(FBchatUserError): + client1.login("", "", max_tries=1) + + client1.login(email, password) + + assert client1.isLoggedIn() + + +@pytest.mark.trylast +def test_sessions(client1): + session = client1.getSession() + Client("no email needed", "no password needed", session_cookies=session) + client1.setSession(session) + assert client1.isLoggedIn() + + +@pytest.mark.tryfirst +def test_default_thread(client1, thread): + client1.setDefaultThread(thread["id"], thread["type"]) + assert client1.send(Message(text="Sent to the specified thread")) + + client1.resetDefaultThread() + with pytest.raises(ValueError): + client1.send(Message(text="Should not be sent")) diff --git a/tests/test_fetch.py b/tests/test_fetch.py new file mode 100644 index 0000000..d755c5a --- /dev/null +++ b/tests/test_fetch.py @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- + +from __future__ import unicode_literals + +import pytest + +from os import path +from fbchat.models import ThreadType, Message, Mention, EmojiSize, Sticker +from utils import subset + + +def test_fetch_all_users(client): + users = client.fetchAllUsers() + assert len(users) > 0 + + +def test_fetch_thread_list(client): + threads = client.fetchThreadList(limit=2) + assert len(threads) == 2 + + +@pytest.mark.parametrize( + "emoji, emoji_size", + [ + ("😆", EmojiSize.SMALL), + ("😆", EmojiSize.MEDIUM), + ("😆", EmojiSize.LARGE), + # These fail because the emoji is made into a sticker + # This should be fixed + pytest.mark.xfail((None, EmojiSize.SMALL)), + pytest.mark.xfail((None, EmojiSize.MEDIUM)), + pytest.mark.xfail((None, EmojiSize.LARGE)), + ], +) +def test_fetch_message_emoji(client, emoji, emoji_size): + mid = client.sendEmoji(emoji, emoji_size) + message, = client.fetchThreadMessages(limit=1) + + assert subset( + vars(message), uid=mid, author=client.uid, text=emoji, emoji_size=emoji_size + ) + + +def test_fetch_message_mentions(client): + text = "This is a test of fetchThreadMessages" + mentions = [Mention(client.uid, offset=10, length=4)] + + mid = client.send(Message(text, mentions=mentions)) + message, = client.fetchThreadMessages(limit=1) + + assert subset(vars(message), uid=mid, author=client.uid, text=text) + for i, m in enumerate(mentions): + assert vars(message.mentions[i]) == vars(m) + + +@pytest.mark.parametrize("sticker_id", ["767334476626295"]) +def test_fetch_message_sticker(client, sticker_id): + mid = client.send(Message(sticker=Sticker(sticker_id))) + message, = client.fetchThreadMessages(limit=1) + + assert subset(vars(message), uid=mid, author=client.uid) + assert subset(vars(message.sticker), uid=sticker_id) + + +def test_fetch_info(client1, group): + info = client1.fetchUserInfo("4")["4"] + assert info.name == "Mark Zuckerberg" + + info = client1.fetchGroupInfo(group["id"])[group["id"]] + assert info.type == ThreadType.GROUP + + +def test_fetch_image_url(client): + url = path.join(path.dirname(__file__), "image.png") + + client.sendLocalImage(url) + message, = client.fetchThreadMessages(limit=1) + + assert client.fetchImageUrl(message.attachments[0].uid) diff --git a/tests/test_message_management.py b/tests/test_message_management.py new file mode 100644 index 0000000..ed6c447 --- /dev/null +++ b/tests/test_message_management.py @@ -0,0 +1,12 @@ +# -*- coding: utf-8 -*- + +from __future__ import unicode_literals + +import pytest + +from fbchat.models import Message, MessageReaction + + +def test_set_reaction(client): + mid = client.send(Message(text="This message will be reacted to")) + client.reactToMessage(mid, MessageReaction.LOVE) diff --git a/tests/test_search.py b/tests/test_search.py new file mode 100644 index 0000000..dda9568 --- /dev/null +++ b/tests/test_search.py @@ -0,0 +1,18 @@ +# -*- coding: utf-8 -*- + +from __future__ import unicode_literals + +from fbchat.models import ThreadType + + +def test_search_for(client1): + users = client1.searchForUsers("Mark Zuckerberg") + assert len(users) > 0 + + u = users[0] + + assert u.uid == "4" + assert u.type == ThreadType.USER + assert u.photo[:4] == "http" + assert u.url[:4] == "http" + assert u.name == "Mark Zuckerberg" diff --git a/tests/test_send.py b/tests/test_send.py new file mode 100644 index 0000000..82098d7 --- /dev/null +++ b/tests/test_send.py @@ -0,0 +1,110 @@ +# -*- coding: utf-8 -*- + +from __future__ import unicode_literals + +import pytest + +from os import path +from fbchat.models import Message, Mention, EmojiSize, FBchatFacebookError, Sticker +from utils import subset + + +@pytest.mark.parametrize( + "text", + [ + "test_send", + "😆", + "\\\n\t%?&'\"", + "ˁҭʚ¹Ʋջوװ՞ޱɣࠚԹБɑȑңКએ֭ʗыԈٌʼőԈ×௴nચϚࠖణٔє܅Ԇޑط", + "a" * 20000, # Maximum amount of characters you can send + ], +) +def test_send_text(client, catch_event, compare, text): + with catch_event("onMessage") as x: + mid = client.sendMessage(text) + + assert compare(x, mid=mid, message=text) + assert subset(vars(x.res["message_object"]), uid=mid, author=client.uid, text=text) + + +@pytest.mark.parametrize( + "emoji, emoji_size", + [ + ("😆", EmojiSize.SMALL), + ("😆", EmojiSize.MEDIUM), + ("😆", EmojiSize.LARGE), + # These fail because the emoji is made into a sticker + # This should be fixed + pytest.mark.xfail((None, EmojiSize.SMALL)), + pytest.mark.xfail((None, EmojiSize.MEDIUM)), + pytest.mark.xfail((None, EmojiSize.LARGE)), + ], +) +def test_send_emoji(client, catch_event, compare, emoji, emoji_size): + with catch_event("onMessage") as x: + mid = client.sendEmoji(emoji, emoji_size) + + assert compare(x, mid=mid, message=emoji) + assert subset( + vars(x.res["message_object"]), + uid=mid, + author=client.uid, + text=emoji, + emoji_size=emoji_size, + ) + + +@pytest.mark.xfail(raises=FBchatFacebookError) +@pytest.mark.parametrize("message", [Message("a" * 20001)]) +def test_send_invalid(client, message): + client.send(message) + + +def test_send_mentions(client, client2, thread, catch_event, compare): + text = "Hi there @me, @other and @thread" + mentions = [ + dict(thread_id=client.uid, offset=9, length=3), + dict(thread_id=client2.uid, offset=14, length=6), + dict(thread_id=thread["id"], offset=26, length=7), + ] + with catch_event("onMessage") as x: + mid = client.send(Message(text, mentions=[Mention(**d) for d in mentions])) + + assert compare(x, mid=mid, message=text) + assert subset(vars(x.res["message_object"]), uid=mid, author=client.uid, text=text) + # The mentions are not ordered by offset + for m in x.res["message_object"].mentions: + assert vars(m) in mentions + + +@pytest.mark.parametrize( + "sticker_id", + ["767334476626295", pytest.mark.xfail("0", raises=FBchatFacebookError)], +) +def test_send_sticker(client, catch_event, compare, sticker_id): + with catch_event("onMessage") as x: + mid = client.send(Message(sticker=Sticker(sticker_id))) + + assert compare(x, mid=mid) + assert subset(vars(x.res["message_object"]), uid=mid, author=client.uid) + assert subset(vars(x.res["message_object"].sticker), uid=sticker_id) + + +@pytest.mark.parametrize( + "method_name, url", + [ + ( + "sendRemoteImage", + "https://github.com/carpedm20/fbchat/raw/master/tests/image.png", + ), + ("sendLocalImage", path.join(path.dirname(__file__), "image.png")), + ], +) +def test_send_images(client, catch_event, compare, method_name, url): + text = "An image sent with {}".format(method_name) + with catch_event("onMessage") as x: + mid = getattr(client, method_name)(url, Message(text)) + + assert compare(x, mid=mid, message=text) + assert subset(vars(x.res["message_object"]), uid=mid, author=client.uid, text=text) + assert x.res["message_object"].attachments[0] diff --git a/tests/test_tests.py b/tests/test_tests.py new file mode 100644 index 0000000..cfb2600 --- /dev/null +++ b/tests/test_tests.py @@ -0,0 +1,12 @@ +# -*- coding: utf-8 -*- + +from __future__ import unicode_literals + +import pytest + + +def test_catch_event(client2, catch_event): + mid = "test" + with catch_event("onMessage") as x: + client2.onMessage(mid=mid) + assert x.res['mid'] == mid diff --git a/tests/test_thread_interraction.py b/tests/test_thread_interraction.py new file mode 100644 index 0000000..7aa744a --- /dev/null +++ b/tests/test_thread_interraction.py @@ -0,0 +1,112 @@ +# -*- coding: utf-8 -*- + +from __future__ import unicode_literals + +import pytest + +from fbchat.models import ( + Message, + ThreadType, + FBchatFacebookError, + TypingStatus, + ThreadColor, +) +from utils import random_hex, subset +from os import environ + + +def test_remove_from_and_add_to_group(client1, client2, group, catch_event): + # Test both methods, while ensuring that the user gets added to the group + try: + with catch_event("onPersonRemoved") as x: + client1.removeUserFromGroup(client2.uid, group["id"]) + assert subset( + x.res, removed_id=client2.uid, author_id=client1.uid, thread_id=group["id"] + ) + finally: + with catch_event("onPeopleAdded") as x: + mid = client1.addUsersToGroup(client2.uid, group["id"]) + assert subset( + x.res, + mid=mid, + added_ids=[client2.uid], + author_id=client1.uid, + thread_id=group["id"], + ) + + +@pytest.mark.xfail( + raises=FBchatFacebookError, reason="Apparently changeThreadTitle is broken" +) +def test_change_title(client1, catch_event, group): + title = random_hex() + with catch_event("onTitleChange") as x: + mid = client1.changeThreadTitle(title, group["id"]) + assert subset( + x.res, + mid=mid, + author_id=client1.uid, + new_title=title, + thread_id=group["id"], + thread_type=ThreadType.GROUP, + ) + + +def test_change_nickname(client, client_all, catch_event, compare): + nickname = random_hex() + with catch_event("onNicknameChange") as x: + client.changeNickname(nickname, client_all.uid) + assert compare(x, changed_for=client_all.uid, new_nickname=nickname) + + +@pytest.mark.parametrize("emoji", ["😀", "😂", "😕", "😍"]) +def test_change_emoji(client, catch_event, compare, emoji): + with catch_event("onEmojiChange") as x: + client.changeThreadEmoji(emoji) + assert compare(x, new_emoji=emoji) + + +@pytest.mark.xfail(raises=FBchatFacebookError) +@pytest.mark.parametrize("emoji", ["🙃", "not an emoji"]) +def test_change_emoji_invalid(client, emoji): + client.changeThreadEmoji(emoji) + + +@pytest.mark.parametrize( + "color", + [ + pytest.mark.xfail( + x, reason="Apparently changing ThreadColor.MESSENGER_BLUE is broken" + ) + if x == ThreadColor.MESSENGER_BLUE + else x + if x == ThreadColor.PUMPKIN + else pytest.mark.expensive(x) + for x in ThreadColor + ], +) +def test_change_color(client, catch_event, compare, color): + if color == ThreadColor.MESSENGER_BLUE: + pytest.xfail(reason="Apparently changing ThreadColor.MESSENGER_BLUE is broken") + with catch_event("onColorChange") as x: + client.changeThreadColor(color) + assert compare(x, new_color=color) + + +@pytest.mark.xfail( + raises=FBchatFacebookError, strict=False, reason="Should fail, but doesn't" +) +def test_change_color_invalid(client): + class InvalidColor: + value = "#0077ff" + + client.changeThreadColor(InvalidColor()) + + +@pytest.mark.xfail(reason="Apparently onTyping is broken") +@pytest.mark.parametrize("status", TypingStatus) +def test_typing_status(client, catch_event, compare, status): + with catch_event("onTyping") as x: + client.setTypingStatus(status) + # x.wait(40) + assert compare(x, status=status) diff --git a/tests/utils.py b/tests/utils.py new file mode 100644 index 0000000..741f798 --- /dev/null +++ b/tests/utils.py @@ -0,0 +1,83 @@ +# -*- coding: utf-8 -*- + +from __future__ import unicode_literals + +import threading +import logging +import six + +from os import environ +from random import randrange +from contextlib import contextmanager +from six import viewitems +from fbchat import Client +from fbchat.models import ThreadType + +log = logging.getLogger("fbchat.tests").addHandler(logging.NullHandler()) + + +class ClientThread(threading.Thread): + def __init__(self, client, *args, **kwargs): + self.client = client + self.should_stop = threading.Event() + super(ClientThread, self).__init__(*args, **kwargs) + + def start(self): + self.client.startListening() + self.client.doOneListen() # QPrimer, Facebook now knows we're about to start pulling + super(ClientThread, self).start() + + def run(self): + while not self.should_stop.is_set() and self.client.doOneListen(): + pass + + self.client.stopListening() + + +if six.PY2: + event_class = threading._Event +else: + event_class = threading.Event + + +class CaughtValue(event_class): + def set(self, res): + self.res = res + super(CaughtValue, self).set() + + def wait(self, timeout=3): + super(CaughtValue, self).wait(timeout=timeout) + + +def random_hex(length=20): + return "{:X}".format(randrange(16 ** length)) + + +def subset(a, **b): + print(a) + print(b) + return viewitems(b) <= viewitems(a) + + +def load_variable(name, cache): + var = environ.get(name, None) + if var is not None: + if cache.get(name, None) != var: + cache.set(name, var) + return var + + var = cache.get(name, None) + if var is None: + raise ValueError("Variable {!r} neither in environment nor cache".format(name)) + return var + + +@contextmanager +def load_client(n, cache): + client = Client( + load_variable("client{}_email".format(n), cache), + load_variable("client{}_password".format(n), cache), + session_cookies=cache.get("client{}_session".format(n), None), + ) + yield client + cache.set("client{}_session".format(n), client.getSession())