From 67afb0acfd25cf2819fc075c473afc99d3601419 Mon Sep 17 00:00:00 2001
From: jrybicki-jsc <j.rybicki@fz-juelich.de>
Date: Tue, 8 Jun 2021 07:48:48 +0200
Subject: [PATCH] authenticte and current user tests
---
apiserver/security/user.py | 19 +++++---------
tests/apiserver_tests/test_responsiveness.py | 8 ++++++
tests/user_tests/test_user.py | 27 ++++++++++++++++----
3 files changed, 36 insertions(+), 18 deletions(-)
diff --git a/apiserver/security/user.py b/apiserver/security/user.py
index 7b106c3..a8880fc 100644
--- a/apiserver/security/user.py
+++ b/apiserver/security/user.py
@@ -114,25 +114,18 @@ def get_password_hash(password):
def authenticate_user(userdb: AbstractDBInterface, username: str, password: str):
- user: UserInDB = get_user(userdb, username)
+ user: UserInDB = userdb.get(username)
if user and verify_password(password, user.hashed_password):
return user
return None
-def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = timedelta(minutes=15)):
to_encode = data.copy()
- if expires_delta:
- expire = datetime.utcnow() + expires_delta
- else:
- expire = datetime.utcnow() + timedelta(minutes=15)
+ expire = datetime.utcnow() + expires_delta
to_encode.update({"exp": expire})
- encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
- return encoded_jwt
-
-
-def get_user(db: AbstractDBInterface, username: str):
- return db.get(username)
+ return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
@@ -144,7 +137,7 @@ def get_current_user(token: str, userdb: AbstractDBInterface):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
- if (username is None) or ((user:=get_user(userdb, username)) is None):
+ if (username is None) or ((user:=userdb.get(username)) is None):
raise credentials_exception
return user
diff --git a/tests/apiserver_tests/test_responsiveness.py b/tests/apiserver_tests/test_responsiveness.py
index 833ca24..621dba3 100644
--- a/tests/apiserver_tests/test_responsiveness.py
+++ b/tests/apiserver_tests/test_responsiveness.py
@@ -36,3 +36,11 @@ class NonAuthTests(unittest.TestCase):
def test_token(self):
rsp = self.client.post('/token', data={'username': 'foo', 'password': 'bar'})
self.assertEqual(rsp.status_code, 401, 'Ath')
+
+ def test_get_non_existing(self):
+ rsp = self.client.get('/dataset/foo')
+ self.assertEqual(404, rsp.status_code)
+ j = rsp.json()
+ self.assertTrue('message' in j, f"{j} should contain message")
+ self.assertTrue('foo' in j['message'], f"{j} should contain object id (foo)")
+
diff --git a/tests/user_tests/test_user.py b/tests/user_tests/test_user.py
index 6e16dc0..b19e6ad 100644
--- a/tests/user_tests/test_user.py
+++ b/tests/user_tests/test_user.py
@@ -1,12 +1,14 @@
import unittest
-from apiserver.security import User, JsonDBInterface, UserInDB
+from apiserver.security import User, JsonDBInterface, UserInDB, authenticate_user, get_current_user
from apiserver.config import ApiserverSettings
+from fastapi import HTTPException
from collections import namedtuple
import os
import pathlib
import shutil
import random
+from unittest.mock import Mock, patch
class UserTests(unittest.TestCase):
@@ -77,7 +79,22 @@ class UserTests(unittest.TestCase):
self.userdb.add(UserInDB(username=f"user_{n}", email='jo@go.com', hashed_password=f"{random.randint(0,200)}"))
self.assertEqual(len(self.userdb.list()), 25)
-
-
-
-
+ def test_not_authenticate_user(self):
+ mock = Mock(spec=JsonDBInterface)
+ mock.get.return_value = None
+ user = authenticate_user(userdb=mock, username='foo', password='pass')
+ self.assertIsNone(user)
+ mock.get.assert_called_with('foo')
+
+ def test_authenticate_user(self):
+ mock = Mock(spec=JsonDBInterface)
+ mock.get.return_value(UserInDB(username='foo', email='bar@o.w', hashed_password='passed'))
+ with patch('apiserver.security.user.verify_password') as vp:
+ user = authenticate_user(userdb=mock, username='foo', password='passed')
+ self.assertIsNotNone(user)
+ vp.assert_called_once()
+ mock.get.assert_called_once()
+ mock.get.assert_called_with('foo')
+
+ def test_current_user(self):
+ self.assertRaises(HTTPException, get_current_user, 'falsetoken', Mock(spec=JsonDBInterface))
\ No newline at end of file
--
GitLab