skillbox/server/oauth/tests/test_hep_client.py

from datetime import datetime, timedelta
from unittest.mock import patch

import requests
from django.test import TestCase

from core.factories import UserFactory
from oauth.factories import Oauth2TokenFactory
from oauth.hep_client import HepClient, HepClientUnauthorizedException, HepClientNoTokenException, HepClientException
from oauth.models import OAuth2Token
from oauth.tests.test_oauth2token import REFRESH_DATA
from users.licenses import MYSKILLBOX_LICENSES
from users.models import License
from users.tests.mock_hep_data_factory import MockResponse

ISBNS = list(MYSKILLBOX_LICENSES.keys())

STUDENT_ISBN = ISBNS[0]
STUDENT_LICENSE = MYSKILLBOX_LICENSES[STUDENT_ISBN]

TEACHER_ISBN = ISBNS[-1]
TEACHER_LICENSE = MYSKILLBOX_LICENSES[TEACHER_ISBN]


class HepClientTestCases(TestCase):
    def setUp(self):
        self.hep_client = HepClient()
        self.now = datetime.now()

    def _create_token(self):
        user = UserFactory(username="bert")
        token = Oauth2TokenFactory(user=user)
        self.token_dict = token.to_token()
        return self.token_dict

    def test_has_no_valid_product(self):
        products = [
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {},
                'activated': self.now - timedelta(2*TEACHER_LICENSE['duration']),
                'status': 'paid'
            },
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {},
                'activated': self.now - timedelta(3 * TEACHER_LICENSE['duration']),
                'status': 'paid'
            },
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {},
                'activated': self.now - timedelta(4 * TEACHER_LICENSE['duration']),
                'status': 'paid'
            }
        ]

        relevant_product = self.hep_client._get_active_product(products)
        self.assertIsNone(relevant_product)

    def test_has_no_not_completed_product(self):
        products = [
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {},
                'activated': self.now - timedelta(7),
                'status': 'not'
            }
        ]

        relevant_product = self.hep_client._get_active_product(products)
        self.assertIsNone(relevant_product)

    def test_has_valid_product(self):
        products = [
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {
                    'id': 0
                },
                'activated': self.now - timedelta(7),
                'status': 'paid'
            },
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {
                    'id': 1
                },
                'activated': self.now - timedelta(3 * TEACHER_LICENSE['duration']),
                'status': 'paid'
            },
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {
                    'id': 2
                },
                'activated': self.now - timedelta(4 * TEACHER_LICENSE['duration']),
                'status': 'paid'
            }
        ]

        relevant_product = self.hep_client._get_active_product(products)
        self.assertEqual(relevant_product['raw']['id'], 0)

    def test_has_multiple_valid_teacher_products_but_only_one_active(self):
        products = [
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {
                    'id': 0
                },
                'activated': self.now - timedelta(7),
                'status': 'paid'
            },
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {
                    'id': 1
                },
                'activated': self.now - timedelta(3 * TEACHER_LICENSE['duration']),
                'status': 'paid'
            },
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {
                    'id': 2
                },
                'activated': self.now - timedelta(4 * TEACHER_LICENSE['duration']),
                'status': 'paid'
            }
        ]

        relevant_product = self.hep_client._get_active_product(products)
        self.assertEqual(relevant_product['raw']['id'], 0)

    def test_has_valid_student_and_teacher_edition(self):
        products = [
            {
                'license': STUDENT_LICENSE,
                'isbn': STUDENT_ISBN,
                'raw': {
                    'id': 0
                },
                'activated': self.now - timedelta(7),
                'status': 'paid'
            },
            {
                'license': TEACHER_LICENSE,
                'isbn': TEACHER_ISBN,
                'raw': {
                    'id': 1
                },
                'activated': self.now - timedelta(7),
                'status': 'paid'
            }
        ]

        relevant_product = self.hep_client._select_from_teacher_products(products)
        self.assertEqual(relevant_product['raw']['id'], 1)

    def test_product_is_active(self):

        expiry_date = self.now + timedelta(3)

        is_active = License.is_product_active(expiry_date, TEACHER_ISBN)
        self.assertTrue(is_active)

    def test_product_is_not_active(self):

        expiry_date = self.now - timedelta(3 * TEACHER_LICENSE['duration'])

        is_active = License.is_product_active(expiry_date, TEACHER_ISBN)
        self.assertFalse(is_active)

    def test_token_is_not_valid_when_token_and_request_empty(self):

        try:
            self.hep_client._get_valid_token(None, None)
        except HepClientNoTokenException:
            return

        self.fail("HepClientTestCases.test_token_is_not_valid_when_token_and_request_empty: Should throw HepClientUnauthorizedException")

    @patch.object(OAuth2Token, 'is_dict_expired', return_value=True)
    @patch.object(requests, 'post', return_value=MockResponse(400, data={}))
    def test_token_is_expired_and_cannot_be_refreshed_from_api(self, mock_fn1, mock_fn2):
        user = UserFactory(username='housi')
        token = Oauth2TokenFactory(user=user).to_token()

        try:
            self.hep_client._get_valid_token(None, token)
        except HepClientException:
            return

        self.fail("HepClientTestCases.test_token_is_expired_and_cannot_be_refreshed_from_api: Should throw HepClientUnauthorizedException")

    @patch.object(OAuth2Token, 'is_dict_expired', return_value=True)
    @patch.object(requests, 'post', return_value=MockResponse(200, data={}))
    @patch.object(OAuth2Token, 'update_dict_with_refresh_data', return_value=(None, False))
    def test_token_is_expired_and_cannot_be_refreshed(self, mock_fn1, mock_fn2, mock_fn3):
        user = UserFactory(username='housi')
        token = Oauth2TokenFactory(user=user).to_token()

        try:
            self.hep_client._get_valid_token(None, token)
        except HepClientUnauthorizedException:
            return

        self.fail("HepClientTestCases.test_token_is_expired_and_cannot_be_refreshed: Should throw HepClientUnauthorizedException")

    @patch.object(OAuth2Token, 'is_dict_expired', return_value=True)
    @patch.object(requests, 'post', return_value=MockResponse(200, data=REFRESH_DATA))
    def test_can_refresh_token(self, mock_fn1, mock_fn2):
        user = UserFactory(username='housi')
        token = Oauth2TokenFactory(user=user).to_token()

        token_dict = self.hep_client._get_valid_token(None, token)
        self.assertEqual(token_dict['access_token'], REFRESH_DATA['access_token'])
        self.assertEqual(token_dict['refresh_token'], REFRESH_DATA['refresh_token'])