pillar/tests/test_badge_sync.py

import datetime

import requests
import responses

from pillar.tests import AbstractPillarTest

httpmock = responses.RequestsMock()


class AbstractSyncTest(AbstractPillarTest):
    def setUp(self):
        super().setUp()
        self.uid1 = self.create_user(24 * '1')
        self.uid2 = self.create_user(24 * '2')

        # Make sure the users have different auth info.
        with self.app.app_context():
            users_coll = self.app.db('users')
            users_coll.update_one(
                {'_id': self.uid1},
                {'$set': {'auth': [
                    {'provider': 'local', 'user_id': '47', 'token': ''},
                    {'provider': 'blender-id', 'user_id': '1947', 'token': ''},
                ]}})
            users_coll.update_one(
                {'_id': self.uid2},
                {'$set': {'auth': [
                    {'provider': 'blender-id', 'user_id': '4488', 'token': ''},
                    {'provider': 'local', 'user_id': '48', 'token': ''},
                ]}})

        self.create_valid_auth_token(self.uid1, token='find-this-token-uid1',
                                     oauth_scopes=['email', 'badge'])
        self.create_valid_auth_token(self.uid1, token='no-badge-scope',
                                     oauth_scopes=['email'])
        self.create_valid_auth_token(self.uid1, token='expired',
                                     oauth_scopes=['email', 'badge'],
                                     expire_in_days=-1)

        self.create_valid_auth_token(self.uid2, token='find-this-token-uid2',
                                     oauth_scopes=['email', 'badge'])
        self.create_valid_auth_token(self.uid2, token='no-badge-scope',
                                     oauth_scopes=['email'])
        self.create_valid_auth_token(self.uid2, token='expired',
                                     oauth_scopes=['email', 'badge'],
                                     expire_in_days=-1)

        from pillar import badge_sync
        self.sync_user1 = badge_sync.SyncUser(self.uid1, 'find-this-token-uid1', '1947')
        self.sync_user2 = badge_sync.SyncUser(self.uid2, 'find-this-token-uid2', '4488')

    def _update_badge_expiry(self, delta_minutes1, delta_minutes2):
        """Make badges of userN expire in delta_minutesN minutes."""
        from pillar.api.utils import utcnow, remove_private_keys
        now = utcnow()

        # Do the update via Eve so that that flow is covered too.
        with self.app.app_context():
            users_coll = self.app.db('users')
            db_user1 = users_coll.find_one(self.uid1)
            db_user1['badges'] = {
                'html': 'badge for user 1',
                'expires': now + datetime.timedelta(minutes=delta_minutes1)
            }
            r, _, _, status = self.app.put_internal('users',
                                                    remove_private_keys(db_user1),
                                                    _id=self.uid1)
        self.assertEqual(200, status, r)

        with self.app.app_context():
            db_user2 = users_coll.find_one(self.uid2)
            db_user2['badges'] = {
                'html': 'badge for user 2',
                'expires': now + datetime.timedelta(minutes=delta_minutes2)
            }
            r, _, _, status = self.app.put_internal('users',
                                                    remove_private_keys(db_user2),
                                                    _id=self.uid2)
        self.assertEqual(200, status, r)


class FindUsersToSyncTest(AbstractSyncTest):
    def test_no_badge_fetched_yet(self):
        from pillar import badge_sync
        with self.app.app_context():
            found = set(badge_sync.find_users_to_sync())
        self.assertEqual({self.sync_user1, self.sync_user2}, found)

    def test_badge_fetched_recently(self):
        from pillar import badge_sync

        # Badges of user1 expired, user2 didn't yet.
        with self.app.app_context():
            self._update_badge_expiry(-5, 5)
            found = list(badge_sync.find_users_to_sync())
        self.assertEqual([self.sync_user1], found)

        # Badges of both users expired, but user2 expired longer ago.
        with self.app.app_context():
            self._update_badge_expiry(-5, -10)
            found = list(badge_sync.find_users_to_sync())
        self.assertEqual([self.sync_user2, self.sync_user1], found)

        # Badges of both not expired yet.
        with self.app.app_context():
            self._update_badge_expiry(2, 3)
            found = list(badge_sync.find_users_to_sync())
        self.assertEqual([], found)


class FindUserSyncInfoTest(AbstractSyncTest):
    def test_no_badge_fetched_yet(self):
        from pillar import badge_sync

        with self.app.app_context():
            found1 = badge_sync.find_user_to_sync(self.uid1)
            found2 = badge_sync.find_user_to_sync(self.uid2)
        self.assertEqual(self.sync_user1, found1)
        self.assertEqual(self.sync_user2, found2)

    def test_badge_fetched_recently(self):
        # This should be the same for all cases, as a single user
        # is always refreshed.
        from pillar import badge_sync

        # Badges of user1 expired, user2 didn't yet.
        with self.app.app_context():
            self._update_badge_expiry(-5, 5)
            found1 = badge_sync.find_user_to_sync(self.uid1)
            found2 = badge_sync.find_user_to_sync(self.uid2)
        self.assertEqual(self.sync_user1, found1)
        self.assertEqual(self.sync_user2, found2)

        # Badges of both users expired, but user2 expired longer ago.
        with self.app.app_context():
            self._update_badge_expiry(-5, -10)
            found1 = badge_sync.find_user_to_sync(self.uid1)
            found2 = badge_sync.find_user_to_sync(self.uid2)
        self.assertEqual(self.sync_user1, found1)
        self.assertEqual(self.sync_user2, found2)

        # Badges of both not expired yet.
        with self.app.app_context():
            self._update_badge_expiry(2, 3)
            found1 = badge_sync.find_user_to_sync(self.uid1)
            found2 = badge_sync.find_user_to_sync(self.uid2)
        self.assertEqual(self.sync_user1, found1)
        self.assertEqual(self.sync_user2, found2)


class FetchHTMLTest(AbstractSyncTest):
    @httpmock.activate
    def test_happy(self):
        from pillar import badge_sync

        def check_request(request: requests.PreparedRequest):
            if request.headers['Authorization'] != 'Bearer find-this-token-uid1':
                return 403, {}, 'BAD TOKEN'
            return 200, {'Content-Type': 'text/html; charset=utf-8'}, 'твоја мајка'.encode()

        httpmock.add_callback('GET', 'http://id.local:8001/api/badges/1947/html/s', check_request)

        with self.app.app_context():
            badge_html = badge_sync.fetch_badge_html(requests.Session(), self.sync_user1, 's')
        self.assertEqual('твоја мајка', badge_html)

    @httpmock.activate
    def test_internal_server_error(self):
        from pillar import badge_sync

        httpmock.add('GET', 'http://id.local:8001/api/badges/1947/html/s',
                     body='oops', status=500)

        with self.assertRaises(badge_sync.StopRefreshing), self.app.app_context():
            badge_sync.fetch_badge_html(requests.Session(), self.sync_user1, 's')

    @httpmock.activate
    def test_no_badge(self):
        from pillar import badge_sync

        httpmock.add('GET', 'http://id.local:8001/api/badges/1947/html/s',
                     body='', status=204)
        with self.app.app_context():
            badge_html = badge_sync.fetch_badge_html(requests.Session(), self.sync_user1, 's')
        self.assertEqual('', badge_html)

    @httpmock.activate
    def test_no_such_user(self):
        from pillar import badge_sync

        httpmock.add('GET', 'http://id.local:8001/api/badges/1947/html/s',
                     body='Not Found', status=404)
        with self.app.app_context():
            badge_html = badge_sync.fetch_badge_html(requests.Session(), self.sync_user1, 's')
        self.assertEqual('', badge_html)

    @httpmock.activate
    def test_no_connection_possible(self):
        from pillar import badge_sync

        with self.assertRaises(badge_sync.StopRefreshing), self.app.app_context():
            badge_sync.fetch_badge_html(requests.Session(), self.sync_user1, 's')


class RefreshAllTest(AbstractSyncTest):
    @httpmock.activate
    def test_happy(self):
        from pillar import badge_sync

        httpmock.add('GET', 'http://id.local:8001/api/badges/1947/html/s',
                     body='badges for Agent 47')
        httpmock.add('GET', 'http://id.local:8001/api/badges/4488/html/s',
                     body='badges for that other user')

        with self.app.app_context():
            badge_sync.refresh_all_badges(timelimit=datetime.timedelta(seconds=4))

        db_user1 = self.get('/api/users/me', auth_token=self.sync_user1.token).json
        db_user2 = self.get('/api/users/me', auth_token=self.sync_user2.token).json
        self.assertEqual('badges for Agent 47', db_user1['badges']['html'])
        self.assertEqual('badges for that other user', db_user2['badges']['html'])

    @httpmock.activate
    def test_timelimit(self):
        from pillar import badge_sync

        # This shouldn't hit any connection error, because it should immediately
        # hit the time limit, before doing any call to Blender ID.
        with self.app.app_context():
            badge_sync.refresh_all_badges(timelimit=datetime.timedelta(seconds=-4))

    @httpmock.activate
    def test_remove_badges(self):
        from pillar import badge_sync
        from pillar.api.utils import utcnow

        # Make sure the user has a badge before getting the 204 No Content from Blender ID.
        self._update_badge_expiry(-10, 10)

        httpmock.add('GET', 'http://id.local:8001/api/badges/1947/html/s',
                     body='', status=204)

        with self.app.app_context():
            badge_sync.refresh_all_badges(
                only_user_id=self.uid1,
                timelimit=datetime.timedelta(seconds=4))
            expected_expire = utcnow() + self.app.config['BLENDER_ID_BADGE_EXPIRY']

            # Get directly from MongoDB because JSON doesn't support datetimes.
            users_coll = self.app.db('users')
            db_user1 = users_coll.find_one(self.uid1)

        self.assertEqual('', db_user1['badges']['html'])
        margin = datetime.timedelta(minutes=1)
        self.assertLess(expected_expire - margin, db_user1['badges']['expires'])
        self.assertGreater(expected_expire + margin, db_user1['badges']['expires'])


class RefreshSingleTest(AbstractSyncTest):
    @httpmock.activate
    def test_happy(self):
        from pillar import badge_sync

        httpmock.add('GET', 'http://id.local:8001/api/badges/1947/html/s',
                     body='badges for Agent 47')

        db_user1 = self.get('/api/users/me', auth_token=self.sync_user1.token).json
        self.assertNotIn('badges', db_user1)

        with self.app.app_context():
            badge_sync.refresh_single_user(self.uid1)

        db_user1 = self.get('/api/users/me', auth_token=self.sync_user1.token).json
        self.assertEqual('badges for Agent 47', db_user1['badges']['html'])