ramon
/
skillbox
1
0
Fork 0
Code Issues 1 Pull Requests 1 Packages Projects 1 Releases 1 Wiki Activity

Add feedback, update tests

This commit is contained in:
Christian Cueni 2021-06-07 08:47:50 +02:00
parent a7ff7c2697
commit 0dc699baf0
27 changed files with 20 additions and 552 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
server/api/schema_public.py
View File

@ -4,10 +4,9 @@ from graphene_django.debug import DjangoDebug
from news.schema_public import AllNewsTeasersQuery
from users.mutations_public import UserMutations
from registration.mutations_public import RegistrationMutations
class PublicMutation(UserMutations, RegistrationMutations, graphene.ObjectType):
class PublicMutation(UserMutations, graphene.ObjectType):
if settings.DEBUG:
debug = graphene.Field(DjangoDebug, name='_debug')

1
server/core/factories.py
View File

@ -55,6 +55,7 @@ class UserFactory(factory.django.DjangoModelFactory):
first_name = factory.LazyAttribute(lambda x: fake.first_name())
last_name = factory.LazyAttribute(lambda x: fake.last_name())
email = factory.LazyAttribute(lambda x: fake.ascii_safe_email())
hep_id = factory.Sequence(lambda n: n)
@factory.post_generation
def post(self, create, extracted, **kwargs):

14
server/core/management/commands/update_admin_token.py
View File

@ -1,14 +0,0 @@
import os
import shutil
from django.core.management import BaseCommand
from core.models import AdminData
class Command(BaseCommand):
def handle(self, *args, **options):
"Update admin token via cronjob"
AdminData.objects.update_admin_token()

15
server/core/managers.py
View File

@ -1,15 +0,0 @@
from django.conf import settings
from django.db import models
from datetime import timedelta
from django.utils import timezone
DEFAULT_PK = 1
class AdminDataManager(models.Manager):
def update_admin_token(self):
return ''
def get_admin_token(self):
return ''

10
server/core/models.py
View File

@ -1,10 +0,0 @@
from django.db import models
from core.managers import AdminDataManager
class AdminData(models.Model):
hep_admin_token = models.CharField(max_length=100, blank=False, null=False)
updated_at = models.DateTimeField(blank=False, null=True, auto_now=True)
objects = AdminDataManager()

9
server/core/utils.py
View File

@ -28,17 +28,14 @@ def is_private_api_call_allowed(user, body):
# logged in users should only be able to access all resources if they have a valid license
# logged in users without valid license have only access to logout, me & coupon mutations
if user.is_anonymous:
return False
if user.is_superuser:
return True
body_unicode = body.decode('utf-8')
try:
if not user.hep_id:
return False
except AttributeError:
return False
if is_endpoint_allowed(body_unicode):
return True

2
server/core/views.py
View File

@ -7,8 +7,6 @@ from django.views.decorators.csrf import ensure_csrf_cookie
from django.views.generic import TemplateView
from graphene_django.views import GraphQLView
from core.models import AdminData
class PrivateGraphQLView(LoginRequiredMixin, GraphQLView):
pass

21
server/oauth/hep_client.py
View File

@ -152,9 +152,7 @@ class HepClient:
for eorder in eorders:
status = ''
if 'status' in eorder:
status = eorder['status']
status = eorder.get('status', '')
for entry in eorder['entries']:
product = self.entry_to_product(entry, self._get_item_activation(eorder), status)
@ -165,7 +163,7 @@ class HepClient:
return products
def entry_to_product(self, entry, activation_date, status):
if is_myskillbox_product(entry['isbn']):
if is_myskillbox_product(entry['isbn']) and activation_date:
return {
'raw': entry,
'activated': activation_date,
@ -180,6 +178,8 @@ class HepClient:
def _get_item_activation(self, eorder):
if 'created_at' in eorder:
return parse_datetime(eorder['created_at'])
else:
return None
def _get_active_product(self, products):
@ -190,10 +190,7 @@ class HepClient:
expiry_delta = product['activated'] + timedelta(product['license']['duration'])
if License.is_product_active(expiry_delta, product['isbn']):
return True
else:
return False
return License.is_product_active(expiry_delta, product['isbn'])
active_products = list(filter(filter_valid_products, products))
@ -205,16 +202,12 @@ class HepClient:
return self._select_from_teacher_products(active_products)
def _select_from_teacher_products(self, active_products):
teacher_edition = None
# select first teacher product, as they are all valid it does not matter which one
for product in active_products:
if product['license']['edition'] == TEACHER_KEY:
teacher_edition = product
break
return product
# select a student product, as they are all valid it does not matter which one
if not teacher_edition:
return active_products[0]
return active_products[0]
return teacher_edition

3
server/oauth/managers.py
View File

@ -18,7 +18,6 @@ class OAuth2TokenManager(models.Manager):
return self._create_oauthtoken(user, token_properties)
def _create_oauthtoken(self, user, token_properties):
token = self.model(user=user, **token_properties)
token.save()
token = self.create(user=user, **token_properties)
return token

6
server/oauth/tests/test_coupon.py
View File

@ -94,8 +94,8 @@ class CouponTests(TestCase):
def test_user_has_valid_skillbox_coupon(self, 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)
user_role = self.user.user_roles.get(user=self.user)
self.assertEqual(user_role.role.key, Role.objects.TEACHER_KEY)
license = License.objects.get(licensee=self.user)
self.assertIsNotNone(license)
@ -110,7 +110,7 @@ class CouponTests(TestCase):
result = self.make_coupon_mutation('COUPON--1234', self.client)
try:
self.user.user_roles.get(user=self.user).role.key
self.user.user_roles.get(user=self.user)
self.fail("CouponTests.test_user_has_valid_non_skillbox_coupon: Should not have created user role")
except:
pass

7
server/oauth/tests/test_middleware.py
View File

@ -1,5 +1,6 @@
from datetime import timedelta
from django.contrib.auth import get_user_model
from django.test import TestCase
from django.utils import timezone
@ -10,9 +11,7 @@ from core.utils import is_private_api_call_allowed
class MiddlewareTestCase(TestCase):
def test_user_without_hep_id_cannot_see_private_api(self):
tomorrow = timezone.now() + timedelta(1)
user = UserFactory(username='aschiman@ch.ch')
user.license_expiry_date = tomorrow
user = get_user_model().objects.create_user(username='sme')
body = b'"{mutation {\\n addRoom}"'
@ -21,7 +20,7 @@ class MiddlewareTestCase(TestCase):
def test_user_with_license_can_see_private_api(self):
tomorrow = timezone.now() + timedelta(1)
user = UserFactory(username='aschiman@ch.ch', hep_id=23)
user = UserFactory(username='aschiman@ch.ch')
user.license_expiry_date = tomorrow.date()
body = b'"{mutation {\\n addRoom}"'

1
server/oauth/urls.py
View File

@ -5,5 +5,4 @@ app_name = 'users'
urlpatterns = [
url(r'^login/', views.login, name='login'),
url(r'^callback/', views.authorize, name='authorize')
# url(r'^oauth/callback/', views.authorize, name='authorize')
]

1
server/oauth/user_signup_login_handler.py
View File

@ -1,4 +1,3 @@
from core.models import AdminData
from oauth.hep_client import HepClient, HepClientException
from users.models import License
from users.models import User, UserRole, Role, SchoolClass

8
server/oauth/views.py
View File

@ -33,12 +33,6 @@ def authorize(request):
if status_msg:
return redirect(f'/login-success?state={status_msg}')
# return status_msg
# return cls.return_login_message(status_msg)
return redirect(f'/login-success?state=success')
# def return_login_message(cls, message):
# # if message == EMAIL_NOT_VERIFIED or message == UNKNOWN_ERROR or message == 'invalid_credentials':
# # raise Exception(message)
# return cls(success=True, message=message)

9
server/registration/__init__.py
View File

@ -1,9 +0,0 @@
# -*- coding: utf-8 -*-
#
# ITerativ GmbH
# http://www.iterativ.ch/
#
# Copyright (c) 2019 ITerativ GmbH. All rights reserved.
#
# Created on 2019-10-08
# @author: chrigu <christian.cueni@iterativ.ch>

6
server/registration/apps.py
View File

@ -1,6 +0,0 @@
from django.apps import AppConfig
class UserConfig(AppConfig):
name = 'registration'

45
server/registration/migrations/0001_initial.py
View File

@ -1,45 +0,0 @@
# Generated by Django 2.0.6 on 2019-10-09 09:05
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
('users', '0009_auto_20191009_0905'),
]
operations = [
migrations.CreateModel(
name='License',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
],
),
migrations.CreateModel(
name='LicenseType',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=255, verbose_name='License name')),
('key', models.CharField(max_length=128)),
('active', models.BooleanField(default=False, verbose_name='License active')),
('description', models.TextField(default='', verbose_name='Description')),
('for_role', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='users.Role')),
],
),
migrations.AddField(
model_name='license',
name='license_type',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='registration.LicenseType'),
),
migrations.AddField(
model_name='license',
name='licensee',
field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL),
),
]

18
server/registration/migrations/0002_auto_20191010_0905.py
View File

@ -1,18 +0,0 @@
# Generated by Django 2.0.6 on 2019-10-10 09:05
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('registration', '0001_initial'),
]
operations = [
migrations.AlterField(
model_name='licensetype',
name='key',
field=models.CharField(max_length=128, unique=True),
),
]

41
server/registration/migrations/0003_auto_20200204_1331.py
View File

@ -1,41 +0,0 @@
# Generated by Django 2.0.6 on 2020-02-04 13:31
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
('users', '0010_schoolclass_code'),
('registration', '0002_auto_20191010_0905'),
]
operations = [
migrations.RemoveField(
model_name='licensetype',
name='for_role',
),
migrations.RemoveField(
model_name='license',
name='license_type',
),
migrations.AddField(
model_name='license',
name='expire_date',
field=models.DateField(null=True),
),
migrations.AddField(
model_name='license',
name='for_role',
field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, to='users.Role'),
),
migrations.AddField(
model_name='license',
name='raw',
field=models.TextField(default=''),
),
migrations.DeleteModel(
name='LicenseType',
),
]

24
server/registration/migrations/0004_auto_20200220_1039.py
View File

@ -1,24 +0,0 @@
# Generated by Django 2.1.15 on 2020-02-20 10:39
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('registration', '0003_auto_20200204_1331'),
]
operations = [
migrations.RemoveField(
model_name='license',
name='for_role',
),
migrations.RemoveField(
model_name='license',
name='licensee',
),
migrations.DeleteModel(
name='License',
),
]

0
server/registration/migrations/__init__.py
View File

68
server/registration/mutations_public.py
View File

@ -1,68 +0,0 @@
# -*- coding: utf-8 -*-
#
# ITerativ GmbH
# http://www.iterativ.ch/
#
# Copyright (c) 2019 ITerativ GmbH. All rights reserved.
#
# Created on 2019-10-08
# @author: chrigu <christian.cueni@iterativ.ch>
import graphene
from django.contrib.auth import login
from graphene import relay
# from core.hep_client import HepClient, HepClientException
from core.models import AdminData
from oauth.user_signup_login_handler import handle_user_and_verify_products, UNKNOWN_ERROR, NO_VALID_LICENSE
class Registration(relay.ClientIDMutation):
class Input:
confirmation_key = graphene.String()
user_id = graphene.Int()
success = graphene.Boolean()
message = graphene.String()
@classmethod
def mutate_and_get_payload(cls, root, info, **kwargs):
confirmation_key = kwargs.get('confirmation_key')
user_id = kwargs.get('user_id')
# hep_client = HepClient()
# admin_token = AdminData.objects.get_admin_token()
# try:
# hep_client.customer_activate(confirmation_key, user_id)
# user_data = hep_client.customers_by_id(admin_token, user_id)
# # double check if user has verified his email. If the "confirmation" field is present, the email address
# # is not verified.
# if 'confirmation' in user_data:
# return cls.return_fail_registration_msg('invalid_key')
# except HepClientException:
# return cls.return_fail_registration_msg('unknown_error')
#
# user, status_msg = handle_user_and_verify_products(user_data)
# if user:
# login(info.context, user)
#
# if status_msg:
# if status_msg == NO_VALID_LICENSE:
# return cls(success=True, message=NO_VALID_LICENSE)
# else:
# return cls.return_fail_registration_msg(status_msg)
return cls(success=True, message='success')
@classmethod
def return_fail_registration_msg(cls, message):
if message == UNKNOWN_ERROR:
raise Exception(message)
return cls(success=False, message=message)
class RegistrationMutations:
registration = Registration.Field()

10
server/registration/tests/__init__.py
View File

@ -1,10 +0,0 @@
# -*- coding: utf-8 -*-
#
# ITerativ GmbH
# http://www.iterativ.ch/
#
# Copyright (c) 2019 ITerativ GmbH. All rights reserved.
#
# Created on 2019-10-08
# @author: chrigu <christian.cueni@iterativ.ch>
from django.conf import settings

101
server/registration/tests/test_registration.py
View File

@ -1,101 +0,0 @@
# -*- coding: utf-8 -*-
#
# ITerativ GmbH
# http://www.iterativ.ch/
#
# Copyright (c) 2019 ITerativ GmbH. All rights reserved.
#
# Created on 2019-10-08
# @author: chrigu <christian.cueni@iterativ.ch>
from unittest.mock import patch
from django.contrib.sessions.middleware import SessionMiddleware
from django.test import TestCase, RequestFactory
from graphene.test import Client
from api.schema_public import schema
# from core.hep_client import HepClient
# from core.tests.mock_hep_data_factory import ME_DATA, VALID_TEACHERS_ORDERS
from users.models import License
from users.models import User, Role, SchoolClass
# INVALID_KEY_ME = dict(ME_DATA)
#
# INVALID_KEY_ME['confirmation'] = 'abddddddd'
# class RegistrationTests(TestCase):
# def setUp(self):
# request = RequestFactory().post('/')
#
# Role.objects.create_default_roles()
#
# self.email = 'sepp@skillbox.iterativ.ch'
# self.first_name = 'Sepp'
# self.last_name = 'Feuz'
#
# # adding session
# middleware = SessionMiddleware()
# middleware.process_request(request)
# request.session.save()
# self.client = Client(schema=schema, context_value=request)
#
# def make_register_mutation(self, confirmation_key, user_id):
# mutation = '''
# mutation Registration($input: RegistrationInput!){
# registration(input: $input) {
# success
# message
# }
# }
# '''
#
# return self.client.execute(mutation, variables={
# 'input': {
# 'confirmationKey': confirmation_key,
# 'userId': user_id
# }
# })
#
# @patch.object(HepClient, 'customer_activate', return_value="Response")
# @patch.object(HepClient, 'customers_by_id', return_value=ME_DATA)
# @patch.object(HepClient, 'myskillbox_product_for_customer', return_value=None)
# @patch.object(HepClient, 'fetch_admin_token', return_value=b'"AABBCCDDEE**44566"')
# def test_user_can_register_with_valid_confirmation_key_and_no_license(self, admin_mock, customer_by_id_mock,
# product_mock, customer_mock):
#
# result = self.make_register_mutation('CONFIRMATION_KEY', 1)
#
# self.assertTrue(result.get('data').get('registration').get('success'))
# self.assertEqual(result.get('data').get('registration').get('message'), 'no_valid_license')
#
# @patch.object(HepClient, 'customer_activate', return_value="Response")
# @patch.object(HepClient, 'customers_by_id', return_value=INVALID_KEY_ME)
# @patch.object(HepClient, 'fetch_admin_token', return_value=b'"AABBCCDDEE**44566"')
# def test_user_cannot_register_with_invalid_key(self, admin_mock, confirmation_mock, id_mock):
#
# result = self.make_register_mutation('CONFIRMATION_KEY', 1)
#
# self.assertFalse(result.get('data').get('registration').get('success'))
# self.assertEqual(result.get('data').get('registration').get('message'), 'invalid_key')
#
# @patch.object(HepClient, '_customer_orders', return_value=VALID_TEACHERS_ORDERS)
# @patch.object(HepClient, 'customer_activate', return_value="Response")
# @patch.object(HepClient, 'customers_by_id', return_value=ME_DATA)
# @patch.object(HepClient, 'fetch_admin_token', return_value=b'"AABBCCDDEE**44566"')
# def test_teacher_can_register_with_remote_license(self, admin_mock, id_mock, activate_mock, orders_mock):
# result = self.make_register_mutation('CONFIRMATION_KEY', 1)
#
# user = User.objects.get(email=ME_DATA['email'])
#
# user_role_key = user.user_roles.get(user=user).role.key
# self.assertEqual(user_role_key, Role.objects.TEACHER_KEY)
#
# license = License.objects.get(licensee=user)
# self.assertEqual(license.for_role.key, Role.objects.TEACHER_KEY)
#
# school_class = SchoolClass.objects.get(users__in=[user])
# self.assertIsNotNone(school_class)
#
# self.assertTrue(result.get('data').get('registration').get('success'))
# self.assertTrue(user.is_authenticated)

94
server/registration/tests/test_registration_proxy.py
View File

@ -1,94 +0,0 @@
# -*- coding: utf-8 -*-
#
# ITerativ GmbH
# http://www.iterativ.ch/
#
# Copyright (c) 2020 ITerativ GmbH. All rights reserved.
#
# Created on 25.02.20
# @author: chrigu <christian.cueni@iterativ.ch>
import json
from unittest.mock import patch
import requests
from django.test import TestCase, Client
from django.urls import reverse
RESPONSE = {
'id': 1234,
'confirmation': 'abdc1234',
'firstname': 'Pesche',
'lastname': 'Zubrüti',
'email': 'aschima@ch.ch',
'prefix': 'Herr',
'gender': 1,
'addresses': [
{
'country_id': 'CH',
'street': ['Weg 1'],
'postcode': '1234',
'city': 'Äussere Einöde',
'firstname': 'Pesche',
'lastname': 'Zubrüti',
'prefix': 'Herr',
'default_shipping': True,
'default_billing': True,
}
],
}
DATA = {
'accepted_terms': True,
'customer': {
'firstname': 'Pesche',
'lastname': 'Zubrüti',
'email': 'aschima@ch.ch',
'prefix': 'Herr',
'gender': 1,
'addresses': [
{
'country_id': 'CH',
'street': ['Weg 1'],
'postcode': '1234',
'city': 'Äussere Einöde',
'firstname': 'Pesche',
'lastname': 'Zubrüti',
'prefix': 'Herr',
'default_shipping': True,
'default_billing': True,
}
],
'password': '123454abasfd'
}
}
# class ProxyTest(TestCase):
#
# def setUp(self):
# self.client = Client()
#
# @patch.object(HepClient, 'customer_create', return_value=RESPONSE)
# def test_proxy_filters_confirmation_key(self, create_mock):
#
# response = self.client.post(reverse('api:registration:proxy'), json.dumps(DATA), content_type="application/json")
# found = 'confirmation' in response.json().keys()
# self.assertFalse(found)
#
# @patch.object(requests, 'post', return_value=MockResponse(400,
# data={'message': 'Ein Kunde mit der gleichen E-Mail-Adresse existiert bereits in einer zugeordneten Website.'}))
# def test_handles_400(self, create_mock):
#
# response = self.client.post(reverse('api:registration:proxy'), json.dumps(DATA), content_type="application/json")
# self.assertEquals(response.json()['message'], 'Ein Kunde mit der gleichen E-Mail-Adresse existiert bereits in einer zugeordneten Website.')
#
# def test_requires_accepted_terms(self):
#
# del DATA['accepted_terms']
#
# response = self.client.post(reverse('api:registration:proxy'), json.dumps(DATA), content_type="application/json")
# self.assertEquals(response.status_code, 400)
# self.assertEquals(response.json()['message'], 'Sie müssen hier zustimmen, damit Sie sich registrieren können.')

9
server/registration/urls.py
View File

@ -1,9 +0,0 @@
from django.conf.urls import url
from django.views.decorators.csrf import csrf_exempt
from registration.view import RegistrationProxyView
app_name = 'registration'
urlpatterns = [
url(r'^registration/', csrf_exempt(RegistrationProxyView.as_view()), name="proxy"),
]

46
server/registration/view.py
View File

@ -1,46 +0,0 @@
# -*- coding: utf-8 -*-
#
# ITerativ GmbH
# http://www.iterativ.ch/
#
# Copyright (c) 2020 ITerativ GmbH. All rights reserved.
#
# Created on 25.02.20
# @author: chrigu <christian.cueni@iterativ.ch>
import json
from django.http import JsonResponse
from django.views import View
class RegistrationProxyView(View):
def post(self, request, *args, **kwargs):
data = json.loads(request.body)
if not self.terms_accepted(data):
return JsonResponse(
{
'message': 'Sie müssen hier zustimmen, damit Sie sich registrieren können.'
},
status=400)
data['customer']['group_id'] = 5
try:
hep_data = hep_client.customer_create(data)
except HepClientException as e:
return JsonResponse(e.args[1], status=e.args[0])
response_data = hep_data.copy()
del response_data['confirmation']
return JsonResponse(response_data)
def terms_accepted(self, data):
if 'accepted_terms' in data and data['accepted_terms']:
del data['accepted_terms']
return True
return False