Add coupons, refactor utility functions

This commit is contained in:
Christian Cueni 2020-02-04 08:15:48 +01:00
parent 23028c779b
commit 321163e542
5 changed files with 187 additions and 16 deletions

View File

@ -55,6 +55,8 @@ class HepClient:
response = requests.get(request_url, headers=headers, data=data)
else:
response = requests.get(request_url, headers=headers)
elif method == 'put':
response = requests.put(request_url, data=data)
# Todo handle 401 and most important network errors
if response.status_code == 401:
@ -130,7 +132,12 @@ class HepClient:
return response.json()
def coupon_redeem(self, coupon, customer_id):
response = self._call('/rest/deutsch/V1/coupon/{}/customer/{}'.format(coupon, customer_id), method='put')
try:
response = self._call('/rest/deutsch/V1/coupon/{}/customer/{}'.format(coupon, customer_id), method='put')
except HepClientException as e:
if e.args[0] == 201:
return None
return response
def myskillbox_product_for_customer(self, admin_token, customer_id):

View File

@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
#
# ITerativ GmbH
# http://www.iterativ.ch/
#
# Copyright (c) 2020 ITerativ GmbH. All rights reserved.
#
# Created on 03.02.20
# @author: chrigu <christian.cueni@iterativ.ch>
import graphene
from graphene import relay
from core.hep_client import HepClient, HepClientException
from users.user_signup_login_handler import check_and_create_licenses, create_role_for_user
class CouponError(graphene.ObjectType):
field = graphene.String()
class Coupon(relay.ClientIDMutation):
class Input:
coupon_code_input = graphene.String()
success = graphene.Boolean()
errors = graphene.List(CouponError) # todo: change for consistency
@classmethod
def mutate_and_get_payload(cls, root, info, **kwargs):
coupon_code = kwargs.get('coupon_code_input')
hep_client = HepClient()
try:
hep_id = info.context.user.hep_id
except AttributeError:
return cls(success=False, errors=[{'field': 'not_authenticated'}])
try:
response = hep_client.coupon_redeem(coupon_code, hep_id)
except HepClientException:
return cls(success=False, errors=[{'field': 'unkown_error'}])
if not response:
return cls(success=False, errors=[{'field': 'invalid_code'}])
license, error_msg = check_and_create_licenses(hep_client, info.context.user)
# todo fail if no license
if error_msg:
return info.context.user, error_msg
create_role_for_user(info.context.user, license.for_role.key)
return cls(success=True, errors=[])

View File

@ -7,8 +7,10 @@
#
# Created on 22.10.18
# @author: chrigu <christian.cueni@iterativ.ch>
from core.schema.mutations.coupon import Coupon
from core.schema.mutations.logout import Logout
class CoreMutations(object):
logout = Logout.Field()
coupon = Coupon.Field()

View File

@ -0,0 +1,95 @@
# -*- coding: utf-8 -*-
#
# ITerativ GmbH
# http://www.iterativ.ch/
#
# Copyright (c) 2020 ITerativ GmbH. All rights reserved.
#
# Created on 03.02.20
# @author: chrigu <christian.cueni@iterativ.ch>
from unittest.mock import patch
import requests
from django.contrib.sessions.middleware import SessionMiddleware
from django.test import TestCase, RequestFactory
from graphene.test import Client
from api.schema import schema
from core.factories import UserFactory
from core.hep_client import HepClient
from core.tests.mock_hep_data_factory import MockResponse, ME_DATA, VALID_TEACHERS_ORDERS
from registration.models import License
from users.models import User, Role, SchoolClass
class CouponTests(TestCase):
def setUp(self):
Role.objects.create_default_roles()
self.user = UserFactory(username='aschi@iterativ.ch', email='aschi@iterativ.ch', hep_id=3)
Role.objects.create_default_roles()
self.teacher_role = Role.objects.get_default_teacher_role()
# adding session
request = RequestFactory().post('/')
middleware = SessionMiddleware()
middleware.process_request(request)
request.user = self.user
request.session.save()
self.client = Client(schema=schema, context_value=request)
def make_coupon_mutation(self, coupon_code, client):
mutation = '''
mutation Coupon($input: CouponInput!){
coupon(input: $input) {
success
errors {
field
}
}
}
'''
return client.execute(mutation, variables={
'input': {
'couponCodeInput': coupon_code
}
})
@patch.object(requests, 'put', return_value=MockResponse(200))
@patch.object(HepClient, '_customer_orders', return_value=VALID_TEACHERS_ORDERS)
@patch.object(HepClient, 'fetch_admin_token', return_value={'token': 'AABBCCDDEE**44566'})
def test_user_has_valid_coupon(self, admin_mock, orders_mock, response_mock):
result = self.make_coupon_mutation('COUPON--1234', self.client)
user_role_key = self.user.user_roles.get(user=self.user).role.key
self.assertEqual(user_role_key, Role.objects.TEACHER_KEY)
license = License.objects.get(licensee=self.user)
self.assertIsNotNone(license)
school_class = SchoolClass.objects.get(users__in=[self.user])
self.assertIsNotNone(school_class)
self.assertTrue(result.get('data').get('coupon').get('success'))
self.assertTrue(self.user.is_authenticated)
@patch.object(requests, 'put', return_value=MockResponse(201))
def test_user_has_invalid_coupon(self, response_mock):
result = self.make_coupon_mutation('COUPON--1234', self.client)
self.assertFalse(result.get('data').get('coupon').get('success'))
self.assertEqual(result.get('data').get('coupon').get('errors')[0].get('field'), 'invalid_code')
@patch.object(requests, 'put', return_value=MockResponse(201))
def test_unauthenticated_user_cannot_redeem(self, response_mock):
request = RequestFactory().post('/')
middleware = SessionMiddleware()
middleware.process_request(request)
request.session.save()
client = Client(schema=schema, context_value=request)
result = self.make_coupon_mutation('COUPON--1234', client)
self.assertFalse(result.get('data').get('coupon').get('success'))
self.assertEqual(result.get('data').get('coupon').get('errors')[0].get('field'), 'not_authenticated')

View File

@ -40,25 +40,38 @@ def handle_user_and_verify_products(user_data):
license = License.objects.get(licensee=user)
# Todo how handle invalid license? Cron Job? How to select correct license? Save all licenses? History?
except License.DoesNotExist:
try:
admin_token = AdminData.objects.get_admin_token()
product = hep_client.myskillbox_product_for_customer(admin_token, user.hep_id)
except HepClientException:
return user, UNKNOWN_ERROR
license, error_msg = check_and_create_licenses(hep_client, user)
if product:
license = License.objects.create_license_for_role(user, product['activated'],
product['raw'], product['edition'])
# todo handle no license case
else:
return user, NO_VALID_LICENSE
if error_msg:
return user, error_msg
UserRole.objects.create_role_for_user(user, license.for_role.key)
if license.for_role.key == Role.objects.TEACHER_KEY:
SchoolClass.create_default_group_for_teacher(user)
create_role_for_user(user, license.for_role.key)
if not license.is_valid():
return user, NO_VALID_LICENSE
return user, None
def check_and_create_licenses(hep_client, user):
try:
admin_token = AdminData.objects.get_admin_token()
product = hep_client.myskillbox_product_for_customer(admin_token, user.hep_id)
except HepClientException:
return None, UNKNOWN_ERROR
if product:
license = License.objects.create_license_for_role(user, product['activated'],
product['raw'], product['edition'])
# todo handle no license case
else:
return None, NO_VALID_LICENSE
return license, None
def create_role_for_user(user, role_key):
UserRole.objects.create_role_for_user(user, role_key)
if role_key == Role.objects.TEACHER_KEY:
SchoolClass.create_default_group_for_teacher(user)