From bfb72fc7d22a8742e76ab005e7931eacbe0d1825 Mon Sep 17 00:00:00 2001
From: Primoz Godec
Date: Fri, 25 Mar 2022 12:53:17 +0100
Subject: [PATCH] owtwitter: adapt to author query
---
orangecontrib/text/widgets/owtwitter.py | 39 ++--
.../text/widgets/tests/test_owtwitter.py | 192 +++++++++++++++---
2 files changed, 187 insertions(+), 44 deletions(-)
diff --git a/orangecontrib/text/widgets/owtwitter.py b/orangecontrib/text/widgets/owtwitter.py
index 232f84900..98d53cc5d 100644
--- a/orangecontrib/text/widgets/owtwitter.py
+++ b/orangecontrib/text/widgets/owtwitter.py
@@ -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
@@ -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,
@@ -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"
@@ -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(
@@ -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
diff --git a/orangecontrib/text/widgets/tests/test_owtwitter.py b/orangecontrib/text/widgets/tests/test_owtwitter.py
index 1f0526eeb..2e9969423 100644
--- a/orangecontrib/text/widgets/tests/test_owtwitter.py
+++ b/orangecontrib/text/widgets/tests/test_owtwitter.py
@@ -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())