Commit b4656dfd authored by Daniel Schreiber's avatar Daniel Schreiber

change management commands

* do not use interactive queries, use parameters instead
* add tests
parent 08f46ac0
Pipeline #27530 passed with stage
in 22 seconds
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from urlparse import urljoin
import requests
from django.core.management import BaseCommand
from rest_framework.authtoken.models import Token
from rest_framework.reverse import reverse
from api.models import TokenAuthUser, UserType
......@@ -22,14 +19,14 @@ class Command(BaseCommand):
headers = {'Authorization': 'token ' + str(token)}
if not user.user_type == UserType.TYPE_SYNC:
continue
continue # pragma: no cover
if not user.api_url:
failed.append(user.username)
continue
try:
url = urljoin(user.api_url, reverse("health"))
url = user.remote_reverse("health")
r = requests.get(url, headers=headers, timeout=5)
if r.status_code == 200:
......@@ -39,8 +36,8 @@ class Command(BaseCommand):
except requests.exceptions.ConnectionError:
failed.append(user.username)
print("{ok} passed, {fail} failed.".format(ok=len(successful), fail=len(failed)))
self.stdout.write("{ok} passed, {fail} failed.".format(ok=len(successful), fail=len(failed)))
if failed:
print("\nFailed:")
self.stdout.write("\nFailed:")
for username in failed:
print(username)
self.stdout.write(username)
......@@ -5,38 +5,25 @@ from django.core.management import BaseCommand
from rest_framework.authtoken.models import Token
from api.models import TokenAuthUser
from api.utils.command_utils import query_yes_no
class Command(BaseCommand):
def add_arguments(self, parser):
parser.add_argument('user_id', nargs='*', type=int)
parser.add_argument('username')
parser.add_argument('--token')
def handle(self, *args, **options):
if options['user_id']:
for user_id in options['user_id']:
try:
user = TokenAuthUser.objects.get(id=user_id)
except TokenAuthUser.DoesNotExist:
print("User {id} does not exist\n".format(id=user_id))
continue
token = Token.objects.get(user_id=user_id)
print("User: " + user.username)
print("Current token: " + token.key)
if query_yes_no("Change token?"):
new_token = raw_input("New token:")
if new_token:
token.delete()
token_obj = Token(user_id=user_id, key=new_token)
token_obj.save()
else:
print("\n")
else:
print("Please specify at least one user id whose token you want to change.")
for user in TokenAuthUser.objects.all():
print("{id}: {name}".format(id=user.id, name=user.username))
try:
user = TokenAuthUser.objects.get(username=options['username'])
except TokenAuthUser.DoesNotExist:
self.stdout.write("User {id} does not exist\n".format(id=options['username']))
return
try:
token = Token.objects.get(user=user)
token.delete()
except Token.DoesNotExist: # pragma: no cover
pass
token_obj = Token(user=user, key=options['token'])
token_obj.save()
self.stdout.write("Token changed.")
......@@ -5,17 +5,18 @@ from dateutil.relativedelta import relativedelta
from django.conf import settings
from django.core.management import BaseCommand
from django.utils import timezone
from rest_framework.authtoken.models import Token
from api.models import ServiceProvider, TokenAuthUser, UserType
class Command(BaseCommand):
def handle(self, *args, **options):
name = raw_input("Name:")
if not name:
return
help = "creates a service provider and a matching user in the database"
def add_arguments(self, parser):
parser.add_argument('name', help="Name of the service provider")
def handle(self, *args, **options):
name = options['name']
sp = ServiceProvider.objects.create(name=name,
expiry_date=timezone.now() + relativedelta(years=1),
realm=settings.SERVER_REALM)
......@@ -28,7 +29,5 @@ class Command(BaseCommand):
sp=sp)
sp_user.save()
sp_user_token = Token.objects.get(user_id=sp_user)
print("SP-UUID: {}".format(sp.uuid))
print("SP-User-UUID: {}".format(sp_user_token.key))
self.stdout.write("SP-UUID: {}".format(sp.uuid))
self.stdout.write("Token for Service Provider: {}".format(sp_user.token()))
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import division, print_function, unicode_literals
# -*- coding: utf-8 -*-
from __future__ import unicode_literals, print_function
from django.core.management import call_command
from django.utils.six import StringIO
from django.test import TestCase
import requests
import requests_mock
from api.models import ServiceProvider, TokenAuthUser, UserType
class AuthTestCommandTest(TestCase):
def setUp(self):
self.user = TokenAuthUser.objects.create_superuser(username="b_sync",
email="test@b.edu",
password="b_sync",
user_type=UserType.TYPE_SYNC,
api_url="http://api.b.edu:7001/",
realm="b.edu")
def test_success(self):
out = StringIO()
with requests_mock.Mocker() as m:
m.register_uri("GET", self.user.remote_reverse("health"), status_code=200)
call_command("auth_test", stdout=out)
self.assertTrue(m.called)
self.assertEqual(out.getvalue(), "1 passed, 0 failed.\n")
def test_failed(self):
out = StringIO()
with requests_mock.Mocker() as m:
m.register_uri("GET", self.user.remote_reverse("health"), status_code=500)
call_command("auth_test", stdout=out)
self.assertTrue(m.called)
self.assertEqual(out.getvalue(), "0 passed, 1 failed.\n\nFailed:\nb_sync\n")
def test_connection_failed(self):
out = StringIO()
with requests_mock.Mocker() as m:
m.register_uri("GET", self.user.remote_reverse("health"), exc=requests.exceptions.ConnectTimeout)
call_command("auth_test", stdout=out)
self.assertTrue(m.called)
self.assertEqual(out.getvalue(), "0 passed, 1 failed.\n\nFailed:\nb_sync\n")
def test_misconfigured_user(self):
self.user.api_url = ""
self.user.save()
out = StringIO()
with requests_mock.Mocker() as m:
m.register_uri("GET", self.user.remote_reverse("health"))
call_command("auth_test", stdout=out)
self.assertFalse(m.called)
self.assertEqual(out.getvalue(), "0 passed, 1 failed.\n\nFailed:\nb_sync\n")
class ChangeTokenCommandTest(TestCase):
def setUp(self):
self.user = TokenAuthUser.objects.create_superuser(username="b_sync",
email="test@b.edu",
password="b_sync",
user_type=UserType.TYPE_SYNC,
api_url="http://api.b.edu:7001/",
realm="b.edu")
def test_change(self):
out = StringIO()
token = "1234567890123456789012345678901234567890"
self.assertNotEqual(self.user.token(), token)
call_command("change_token", self.user.username, token=token, stdout=out)
user = TokenAuthUser.objects.get(pk=self.user.pk)
self.assertEqual(user.token(), token)
self.assertEqual(out.getvalue(), "Token changed.\n")
def test_doesn_not_exist(self):
out = StringIO()
token = "1234567890123456789012345678901234567890"
call_command("change_token", "foobar", token=token, stdout=out)
self.assertEqual(out.getvalue(), "User foobar does not exist\n")
class CreateSpCommandTest(TestCase):
def test_create_sp(self):
out = StringIO()
servicename = "foobar"
call_command("create_sp", servicename, stdout=out)
user = TokenAuthUser.objects.get(username="sp_{}".format(servicename))
sp = ServiceProvider.objects.get(name=servicename)
self.assertIn("SP-UUID: {}".format(sp.pk), out.getvalue())
self.assertIn("Token for Service Provider: {}".format(user.token()), out.getvalue())
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment