Skip to content

Commit

Permalink
owtwitter: adapt to author query
Browse files Browse the repository at this point in the history
  • Loading branch information
@PrimozGodec
PrimozGodec committed Mar 25, 2022
1 parent b965281 commit bfb72fc
Show file tree
Hide file tree
Showing 2 changed files with 187 additions and 44 deletions.
39 changes: 26 additions & 13 deletions orangecontrib/text/widgets/owtwitter.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@

from orangecontrib.text import twitter
from orangecontrib.text.corpus import Corpus
from orangecontrib.text.language_codes import lang2code, code2lang
from orangecontrib.text.twitter import TwitterAPI, SUPPORTED_LANGUAGES
from orangecontrib.text.language_codes import code2lang
from orangecontrib.text.twitter import TwitterAPI, SUPPORTED_LANGUAGES, NoAuthorError
from orangecontrib.text.widgets.utils import ComboBox, ListEdit, gui_require


Expand Down Expand Up @@ -42,7 +42,7 @@ def advance(progress):
collecting=collecting,
callback=advance,
)
elif mode == "authors":
else: # mode == "authors":
return api.search_authors(
max_tweets=max_tweets,
authors=word_list,
Expand Down Expand Up @@ -70,19 +70,20 @@ class Info(OWWidget.Information):
)

class Error(OWWidget.Error):
api_error = Msg("Api error ({})")
api_error = Msg("Api error: {}")
empty_query = Msg("Please provide {}.")
key_missing = Msg("Please provide a valid API token.")
wrong_author = Msg("Author '{}' does not exist.")

CONTENT, AUTHOR = 0, 1
MODES = ["Content", "Author"]
word_list = Setting([])
mode = Setting(0)
limited_search = Setting(True)
max_tweets = Setting(100)
language = Setting(None)
allow_retweets = Setting(False)
collecting = Setting(False)
word_list: List = Setting([])
mode: int = Setting(0)
limited_search: bool = Setting(True)
max_tweets: int = Setting(100)
language: Optional[str] = Setting(None)
allow_retweets: bool = Setting(False)
collecting: bool = Setting(False)

class APICredentialsDialog(OWWidget):
name = "Twitter API Credentials"
Expand Down Expand Up @@ -240,6 +241,7 @@ def search(self):
content = self.mode == self.CONTENT
if not self.word_list:
self.Error.empty_query("keywords" if content else "authors")
self.Outputs.corpus.send(None)
return

self.start(
Expand All @@ -263,13 +265,24 @@ def update_api(self, key):

def on_done(self, result_corpus):
self.search_button.setText("Search")
if len(result_corpus) < self.max_tweets:
if (
result_corpus is None # probably because of rate error at beginning
# or fewer tweets than expected
or self.mode == self.CONTENT
and len(result_corpus) < self.max_tweets
or self.mode == self.AUTHOR
# for authors, we expect self.max_tweets for each author
and len(result_corpus) < self.max_tweets * len(self.word_list)
):
self.Info.nut_enough_tweets()
self.Outputs.corpus.send(result_corpus)

def on_exception(self, ex):
self.search_button.setText("Search")
self.Error.api_error(str(ex))
if isinstance(ex, NoAuthorError):
self.Error.wrong_author(str(ex))
else:
self.Error.api_error(str(ex))

def on_partial_result(self, _):
pass
Expand Down
192 changes: 161 additions & 31 deletions orangecontrib/text/widgets/tests/test_owtwitter.py
View file
Original file line number Diff line number Diff line change
@@ -1,66 +1,196 @@
import unittest
from unittest.mock import patch

from unittest.mock import patch, MagicMock

from Orange.widgets.tests.base import WidgetTest
from orangecontrib.text import twitter, Corpus
from Orange.widgets.tests.utils import simulate
from tweepy import TooManyRequests, TweepyException

from orangecontrib.text.tests.test_twitter import (
DummyPaginator,
tweets,
users,
places,
TestTwitterAPI,
)
from orangecontrib.text.widgets.owtwitter import OWTwitter
from tweepy import TweepyException, TooManyRequests


@patch("tweepy.Client.get_user", MagicMock())
class TestTwitterWidget(WidgetTest):
def setUp(self):
self.widget = self.create_widget(OWTwitter)
self.widget: OWTwitter = self.create_widget(OWTwitter)
# give some key to api - to allow start the search
self.widget.update_api(twitter.Credentials("testkey", "testsecret"))

def test_no_error(self):
self.widget.search()
self.assertFalse(self.widget.Error.empty_query.is_shown())
self.widget.update_api("test_key")

def test_empty_author_list(self):
self.widget.mode = 1
self.widget.mode_toggle()
def test_empty_query_error(self):
self.widget.search_button.click()
self.wait_until_finished()
self.assertTrue(self.widget.Error.empty_query.is_shown())
self.assertTrue(str(self.widget.Error.empty_query).endswith("keywords."))
self.assertIsNone(self.get_output(self.widget.Outputs.corpus))

@patch("orangecontrib.text.twitter.TwitterAPI.fetch", dummy_fetch)
def test_content_search(self):
self.widget.word_list = ["orange"]
simulate.combobox_activate_item(self.widget.controls.mode, "Author")
self.widget.search_button.click()
output = self.get_output(self.widget.Outputs.corpus)
self.assertEqual(3, len(output))
self.assertGreater(len(str(output[0, "Content"])), 0)
self.assertTrue(self.widget.Error.empty_query.is_shown())
self.assertTrue(str(self.widget.Error.empty_query).endswith("authors."))
self.assertIsNone(self.get_output(self.widget.Outputs.corpus))

@patch("orangecontrib.text.twitter.TwitterAPI.fetch", dummy_fetch)
@patch("tweepy.Paginator", DummyPaginator(tweets, users, places))
def test_author(self):
self.widget.mode = 1
simulate.combobox_activate_item(self.widget.controls.mode, "Author")
self.widget.word_list = ["@OrangeDataMiner"]
self.widget.mode_toggle()
self.widget.search_button.click()

output = self.get_output(self.widget.Outputs.corpus)
self.assertEqual(3, len(output))
self.assertGreater(len(str(output[0, "Content"])), 0)
self.assertEqual(4, len(output))

@patch("tweepy.Cursor.items")
self.widget.word_list = ["OrangeDataMiner", "test"]
self.widget.search_button.click()

output = self.get_output(self.widget.Outputs.corpus)
self.assertEqual(4, len(output))

self.widget.word_list = []
self.widget.search_button.click()

output = self.get_output(self.widget.Outputs.corpus)
self.assertIsNone(output)

@patch("tweepy.Paginator", DummyPaginator(tweets, users, places))
def test_content(self):
simulate.combobox_activate_item(self.widget.controls.mode, "Content")
self.widget.word_list = ["OrangeDataMiner"]
self.widget.search_button.click()

output = self.get_output(self.widget.Outputs.corpus)
self.assertEqual(4, len(output))

self.widget.word_list = ["OrangeDataMiner", "test"]
self.widget.search_button.click()

output = self.get_output(self.widget.Outputs.corpus)
self.assertEqual(4, len(output))

self.widget.word_list = []
self.widget.search_button.click()

output = self.get_output(self.widget.Outputs.corpus)
self.assertIsNone(output)

@patch("tweepy.Paginator")
def test_rate_limit(self, mock_items):
mock_items.side_effect = TooManyRequests(Response(492))
mock_items.__iter__.side_effect = TooManyRequests(MagicMock())
self.widget.word_list = ["orange"]
self.widget.search_button.click()
self.wait_until_finished()
self.assertTrue(self.widget.Error.rate_limit.is_shown())
self.assertTrue(self.widget.Info.nut_enough_tweets.is_shown())
# since rate error happen at beginning no tweets are download so far
self.assertIsNone(self.get_output(self.widget.Outputs.corpus))
self.assertEqual("Search", self.widget.search_button.text())

@patch("tweepy.Cursor.items")
def test_error(self, mock_items):
mock_items.side_effect = TweepyException("Other errors", Response(400))
@patch("tweepy.Paginator", side_effect=TweepyException("Other"))
def test_tweepy_error(self, _):
self.widget.word_list = ["orange"]
self.widget.search_button.click()
self.wait_until_finished()
self.assertTrue(self.widget.Error.api_error.is_shown())
self.assertEqual("Api error: Other", str(self.widget.Error.api_error))
self.assertEqual("Search", self.widget.search_button.text())

def test_author_not_existing(self):
with patch("tweepy.Client.get_user") as m:
m.return_value = MagicMock(data=None)
simulate.combobox_activate_item(self.widget.controls.mode, "Author")
self.widget.word_list = ["orange"]
self.widget.search_button.click()
self.wait_until_finished()
self.assertTrue(self.widget.Error.wrong_author.is_shown())
self.assertEqual(
"Author 'orange' does not exist.", str(self.widget.Error.wrong_author)
)
self.assertEqual("Search", self.widget.search_button.text())

@patch("tweepy.Paginator")
def test_language(self, mock):
simulate.combobox_activate_item(self.widget.controls.mode, "Content")
simulate.combobox_activate_item(self.widget.language_combo, "English")
self.widget.word_list = ["OrangeDataMiner"]
self.widget.search_button.click()
self.wait_until_finished()

TestTwitterAPI.assert_query(mock, '"OrangeDataMiner" -is:retweet lang:en')
mock.reset_mock()

simulate.combobox_activate_item(self.widget.language_combo, "Slovene")
self.widget.search_button.click()
self.wait_until_finished()

TestTwitterAPI.assert_query(mock, '"OrangeDataMiner" -is:retweet lang:sl')
mock.reset_mock()

simulate.combobox_activate_item(self.widget.language_combo, "German")
self.widget.search_button.click()
self.wait_until_finished()

TestTwitterAPI.assert_query(mock, '"OrangeDataMiner" -is:retweet lang:de')

@patch("tweepy.Paginator")
def test_is_retweet(self, mock):
self.widget.retweets_checkbox.setChecked(False)
self.widget.word_list = ["OrangeDataMiner"]
self.widget.search_button.click()
self.wait_until_finished()

TestTwitterAPI.assert_query(mock, '"OrangeDataMiner" -is:retweet')
mock.reset_mock()

self.widget.retweets_checkbox.setChecked(True)
self.widget.search_button.click()
self.wait_until_finished()

TestTwitterAPI.assert_query(mock, '"OrangeDataMiner"')
mock.reset_mock()

self.widget.retweets_checkbox.setChecked(False)
self.widget.search_button.click()
self.wait_until_finished()

TestTwitterAPI.assert_query(mock, '"OrangeDataMiner" -is:retweet')
mock.reset_mock()

@patch("tweepy.Paginator", DummyPaginator(tweets, users, places))
def test_max_tweets(self):
simulate.combobox_activate_item(self.widget.controls.mode, "Content")
self.widget.controls.max_tweets.setValue(2)
self.widget.word_list = ["OrangeDataMiner"]
self.widget.search_button.click()

output = self.get_output(self.widget.Outputs.corpus)
self.assertEqual(2, len(output))

self.widget.controls.max_tweets.setValue(3)
self.widget.word_list = ["OrangeDataMiner"]
self.widget.search_button.click()

output = self.get_output(self.widget.Outputs.corpus)
self.assertEqual(3, len(output))

@patch("tweepy.Paginator")
def test_send_report(self, _):
simulate.combobox_activate_item(self.widget.controls.mode, "Content")
self.widget.controls.max_tweets.setValue(2)
self.widget.word_list = ["OrangeDataMiner"]
self.widget.search_button.click()
self.wait_until_finished()

self.widget.send_report()

@patch("tweepy.Paginator", DummyPaginator(tweets, users, places))
def test_interrupted(self):
self.widget.word_list = ["OrangeDataMiner"]
self.widget.search_button.click()
self.assertEqual("Stop", self.widget.search_button.text())
self.widget.search_button.click()
self.wait_until_finished()
self.assertEqual("Search", self.widget.search_button.text())


Expand Down

0 comments on commit bfb72fc

Please sign in to comment.