TWITTER_TWEET_NOT_FOUND_ERROR = 144 TWITTER_USER_NOT_FOUND_ERROR = 50 TWITTER_PAGE_DOES_NOT_EXISTS_ERROR = 34 TWITTER_DELETE_OTHER_USER_TWEET = 183 TWITTER_ACCOUNT_SUSPENDED_ERROR = 64 TWITTER_USER_IS_NOT_LIST_MEMBER_SUBSCRIBER = 109 class APIQuotaError(APIError):
    """
    Exception indicate some API quota breached.
    """
    pass class CharacterLimitError(ResponseBotError):
    """
    Error to indicate your tweet reached the character limit.
    """
    pass class AutomatedRequestError(ResponseBotError):
    """
    Error to indicate a request is deemed automated by Twitter.
    """
    pass class OverCapacityError(ResponseBotError):
    """
    Error to indicate Twitter is currently over capacity.
    """
    pass class DailyStatusUpdateError(ResponseBotError):
    """
    Error to indicate your account reached the daily status update limit.
    """
    pass class StatusDuplicateError(ResponseBotError):
    """
    Error to indicate your status is a duplicate.
    """
    pass class ResponseBotClient(object):
    """
    Wrapper for all Twitter API clients.
    """
    def __init__(self, client, config):
        self._client = client
        self._current_user = None
        self.config = config

    @property
    def tweepy_api(self):
        """
        Get the actual client object.

        :return: the actual client object
        """
        return self._client

    def get_current_user(self):
        if self._current_user is None:
            self._current_user = User(self._client.me()._json)
        return self._current_user

    @api_error_handle
    def tweet(self, text, in_reply_to=None, filename=None, file=None):
        """
        Post a new tweet.

        :param text: the text to post
        :param in_reply_to: The ID of the tweet to reply to
        :param filename: If `file` param is not provided, read file from this path
        :param file: A file object, which will be used instead of opening `filename`. `filename` is still required, for
        MIME type detection and to use as a form field in the POST data
        :return: Tweet object
        """

        if filename is None:
            return Tweet(self._client.update_status(status=text, in_reply_to_status_id=in_reply_to)._json)
        else:
            return Tweet(self._client.update_with_media(filename=filename, file=file,
                                                        status=text, in_reply_to_status_id=in_reply_to)._json)

    def retweet(self, id):
        """
        Retweet a tweet.

        :param id: ID of the tweet in question
        :return: True if success, False otherwise
        """
        try:
            self._client.retweet(id=id)
            return True
        except TweepError as e:
            if e.api_code == TWITTER_PAGE_DOES_NOT_EXISTS_ERROR:
                return False
            raise

    def get_tweet(self, id):
        """
        Get an existing tweet.

        :param id: ID of the tweet in question
        :return: Tweet object. None if not found
        """
        try:
            return Tweet(self._client.get_status(id=id)._json)
        except TweepError as e:
            if e.api_code == TWITTER_TWEET_NOT_FOUND_ERROR:
                return None
            raise

    def get_user(self, id):
        """
        Get a user's info.

        :param id: ID of the user in question
        :return: User object. None if not found
        """
        try:
            return User(self._client.get_user(user_id=id)._json)
        except TweepError as e:
            if e.api_code == TWITTER_USER_NOT_FOUND_ERROR:
                return None
            raise

    def remove_tweet(self, id):
        """
        Delete a tweet.

        :param id: ID of the tweet in question
        :return: True if success, False otherwise
        """
        try:
            self._client.destroy_status(id=id)
            return True
        except TweepError as e:
            if e.api_code in [TWITTER_PAGE_DOES_NOT_EXISTS_ERROR, TWITTER_DELETE_OTHER_USER_TWEET]:
                return False
            raise

    def follow(self, user_id, notify=False):
        """
        Follow a user.

        :param user_id: ID of the user in question
        :param notify: whether to notify the user about the following
        :return: user that are followed
        """
        try:
            return User(self._client.create_friendship(user_id=user_id, follow=notify)._json)
        except TweepError as e:
            if e.api_code in [TWITTER_ACCOUNT_SUSPENDED_ERROR]:
                return self.get_user(user_id)
            raise

    def unfollow(self, user_id):
        """
        Follow a user.

        :param user_id: ID of the user in question
        :return: The user that were unfollowed
        """
        return User(self._client.destroy_friendship(user_id=user_id)._json)

    ###################################################################################
    # Lists
    ###################################################################################
    @api_error_handle
    def create_list(self, name, mode='public', description=None):
        """
        Create a list

        :param name: Name of the new list
        :param mode: :code:`'public'` (default) or :code:`'private'`
        :param description: Description of the new list
        :return: The new list object
        :rtype: :class:`~responsebot.models.List`
        """
        return List(tweepy_list_to_json(self._client.create_list(name=name, mode=mode, description=description)))

    @api_error_handle
    def destroy_list(self, list_id):
        """
        Destroy a list

        :param list_id: list ID number
        :return: The destroyed list object
        :rtype: :class:`~responsebot.models.List`
        """
        return List(tweepy_list_to_json(self._client.destroy_list(list_id=list_id)))

    @api_error_handle
    def update_list(self, list_id, name=None, mode=None, description=None):
        """
        Update a list

        :param list_id: list ID number
        :param name: New name for the list
        :param mode: :code:`'public'` (default) or :code:`'private'`
        :param description: New description of the list
        :return: The updated list object
        :rtype: :class:`~responsebot.models.List`
        """
        return List(tweepy_list_to_json(
            self._client.update_list(list_id=list_id, name=name, mode=mode, description=description))
        )

    @api_error_handle
    def lists(self):
        """
        List user's lists

        :return: list of :class:`~responsebot.models.List` objects
        """
        return [List(tweepy_list_to_json(list)) for list in self._client.lists_all()]

    @api_error_handle
    def lists_memberships(self):
        """
        List lists which user was added

        :return: list of :class:`~responsebot.models.List` objects
        """
        return [List(tweepy_list_to_json(list)) for list in self._client.lists_memberships()]

    @api_error_handle
    def lists_subscriptions(self):
        """
        List lists which user followed

        :return: list of :class:`~responsebot.models.List` objects
        """
        return [List(tweepy_list_to_json(list)) for list in self._client.lists_subscriptions()]

    @api_error_handle
    def list_timeline(self, list_id, since_id=None, max_id=None, count=20):
        """
        List the tweets of specified list.

        :param list_id: list ID number
        :param since_id: results will have ID greater than specified ID (more recent than)
        :param max_id: results will have ID less than specified ID (older than)
        :param count: number of results per page
        :return: list of :class:`~responsebot.models.Tweet` objects
        """
        statuses = self._client.list_timeline(list_id=list_id, since_id=since_id, max_id=max_id, count=count)
        return [Tweet(tweet._json) for tweet in statuses]

    @api_error_handle
    def get_list(self, list_id):
        """
        Get info of specified list

        :param list_id: list ID number
        :return: :class:`~responsebot.models.List` object
        """
        return List(tweepy_list_to_json(self._client.get_list(list_id=list_id)))

    @api_error_handle
    def add_list_member(self, list_id, user_id):
        """
        Add a user to list

        :param list_id: list ID number
        :param user_id: user ID number
        :return: :class:`~responsebot.models.List` object
        """
        return List(tweepy_list_to_json(self._client.add_list_member(list_id=list_id, user_id=user_id)))

    @api_error_handle
    def remove_list_member(self, list_id, user_id):
        """
        Remove a user from a list

        :param list_id: list ID number
        :param user_id: user ID number
        :return: :class:`~responsebot.models.List` object
        """
        return List(tweepy_list_to_json(self._client.remove_list_member(list_id=list_id, user_id=user_id)))

    @api_error_handle
    def list_members(self, list_id):
        """
        List users in a list

        :param list_id: list ID number
        :return: list of :class:`~responsebot.models.User` objects
        """
        return [User(user._json) for user in self._client.list_members(list_id=list_id)]

    @api_error_handle
    def is_list_member(self, list_id, user_id):
        """
        Check if a user is member of a list

        :param list_id: list ID number
        :param user_id: user ID number
        :return: :code:`True` if user is member of list, :code:`False` otherwise
        """
        try:
            return bool(self._client.show_list_member(list_id=list_id, user_id=user_id))
        except TweepError as e:
            if e.api_code == TWITTER_USER_IS_NOT_LIST_MEMBER_SUBSCRIBER:
                return False
            raise

    @api_error_handle
    def subscribe_list(self, list_id):
        """
        Subscribe to a list

        :param list_id: list ID number
        :return: :class:`~responsebot.models.List` object
        """
        return List(tweepy_list_to_json(self._client.subscribe_list(list_id=list_id)))

    @api_error_handle
    def unsubscribe_list(self, list_id):
        """
        Unsubscribe to a list

        :param list_id: list ID number
        :return: :class:`~responsebot.models.List` object
        """
        return List(tweepy_list_to_json(self._client.unsubscribe_list(list_id=list_id)))

    @api_error_handle
    def list_subscribers(self, list_id):
        """
        List subscribers of a list

        :param list_id: list ID number
        :return: :class:`~responsebot.models.User` object
        """
        return [User(user._json) for user in self._client.list_subscribers(list_id=list_id)]

    @api_error_handle
    def is_subscribed_list(self, list_id, user_id):
        """
        Check if user is a subscribed of specified list

        :param list_id: list ID number
        :param user_id: user ID number
        :return: :code:`True` if user is subscribed of list, :code:`False` otherwise
        """
        try:
            return bool(self._client.show_list_subscriber(list_id=list_id, user_id=user_id))
        except TweepError as e:
            if e.api_code == TWITTER_USER_IS_NOT_LIST_MEMBER_SUBSCRIBER:
                return False
            raise
from unittest.case import TestCase
from datetime import datetime
from tweepy.error import TweepError, RateLimitError
from responsebot.common.constants import TWITTER_TWEET_NOT_FOUND_ERROR, TWITTER_USER_NOT_FOUND_ERROR, \
    TWITTER_PAGE_DOES_NOT_EXISTS_ERROR, TWITTER_DELETE_OTHER_USER_TWEET, TWITTER_ACCOUNT_SUSPENDED_ERROR, \
    TWITTER_USER_IS_NOT_LIST_MEMBER_SUBSCRIBER
from responsebot.common.exceptions import APIQuotaError, CharacterLimitError, AutomatedRequestError,\
    OverCapacityError, DailyStatusUpdateError, StatusDuplicateError
from responsebot.responsebot_client import ResponseBotClient
    from mock import MagicMock
    from unittest.mock import MagicMock



try:
except ImportError:


class TweetClientTestCase(TestCase):
    def setUp(self):
        self.real_client = MagicMock()
        self.config = MagicMock()
        self.client = ResponseBotClient(self.real_client, self.config)

    def test_post_new_tweet(self):
        self.real_client.update_status = MagicMock(return_value=MagicMock(_json={
            'some_key': 'some value',
        }))

        tweet = self.client.tweet('some text')

        self.real_client.update_status.assert_called_once_with(status='some text', in_reply_to_status_id=None)
        self.real_client.update_with_media.assert_not_called()
        self.assertEqual(tweet.some_key, 'some value')

    def test_reply(self):
        self.real_client.update_status = MagicMock(return_value=MagicMock(_json={
            'some_key': 'some value',
        }))

        tweet = self.client.tweet('some text', in_reply_to=123)

        self.real_client.update_status.assert_called_once_with(status='some text', in_reply_to_status_id=123)
        self.assertEqual(tweet.some_key, 'some value')

    def test_post_new_tweet_exceed_character_limit(self):
        self.real_client.update_status = MagicMock(
            side_effect=TweepError(api_code=186, reason='Status is more than 140 characters.'))

        self.assertRaises(
            CharacterLimitError,
            self.client.tweet,
            text="some text with more than Twitter's character limit"
        )

    def test_post_new_tweet_automated_error(self):
        self.real_client.update_status = MagicMock(
            side_effect=TweepError(api_code=226, reason='Automated'))

        self.assertRaises(
            AutomatedRequestError,
            self.client.tweet,
            text="some text with more than Twitter's character limit"
        )

    def test_post_new_tweet_twitter_over_capacity(self):
        self.real_client.update_status = MagicMock(
            side_effect=TweepError(api_code=130, reason='Over capacity'))

        self.assertRaises(
            OverCapacityError,
            self.client.tweet,
            text="some text with more than Twitter's character limit"
        )

    def test_post_new_tweet_hit_daily_update_limit(self):
        self.real_client.update_status = MagicMock(
            side_effect=TweepError(api_code=185, reason='Daily update limit reached'))

        self.assertRaises(

            DailyStatusUpdateError,