api.auth: start with simple dummy (testing) authentication
This commit is contained in:
parent
fc0616ce6a
commit
a139fd784e
3 changed files with 15 additions and 8 deletions
|
|
@ -46,7 +46,7 @@ def getUser(storage: Storage, userName: str) -> User:
|
||||||
user = users.queryLast(name=userName)
|
user = users.queryLast(name=userName)
|
||||||
return user.toObject(User)
|
return user.toObject(User)
|
||||||
|
|
||||||
async def authenticate(storage: deps.Storage, token: deps.Auth):
|
async def authenticate(storage: deps.Storage, token: deps.AuthUser):
|
||||||
payload = jwt.decode(token, settings.jwt_key, algorithms=[settings.jwt_algorithm])
|
payload = jwt.decode(token, settings.jwt_key, algorithms=[settings.jwt_algorithm])
|
||||||
userName = payload.get('sub')
|
userName = payload.get('sub')
|
||||||
user = getUser(storage, userName)
|
user = getUser(storage, userName)
|
||||||
|
|
@ -63,7 +63,9 @@ async def get_users(storage: deps.Storage):
|
||||||
return result
|
return result
|
||||||
|
|
||||||
@router.post('/users', response_model=User)
|
@router.post('/users', response_model=User)
|
||||||
async def post_user(user: Annotated[UserIn, Body()], storage: deps.Storage):
|
async def post_user(user: Annotated[UserIn, Body()], storage: deps.Storage,
|
||||||
|
auth_user: deps.AuthUser):
|
||||||
|
print('*** post_user', auth_user)
|
||||||
users = storage.create(st_user.Users)
|
users = storage.create(st_user.Users)
|
||||||
tr = st_user.User.fromObject(user)
|
tr = st_user.User.fromObject(user)
|
||||||
users.save(tr)
|
users.save(tr)
|
||||||
|
|
|
||||||
|
|
@ -10,13 +10,14 @@ from scopes.api.auth import config as auth_config
|
||||||
from scopes.storage import common
|
from scopes.storage import common
|
||||||
|
|
||||||
type Storage = Annotated[common.Storage, Depends(storage)]
|
type Storage = Annotated[common.Storage, Depends(storage)]
|
||||||
type Auth = Annotated[str, Depends(auth_method())]
|
type AuthUser = Annotated[str, Depends(auth_user)]
|
||||||
|
|
||||||
def storage():
|
def storage():
|
||||||
stf = config.StorageFactory(config.settings)
|
stf = config.StorageFactory(config.settings)
|
||||||
return stf(config.settings.dbschema)
|
return stf(config.settings.dbschema)
|
||||||
|
|
||||||
def auth_method():
|
def auth_user():
|
||||||
auth = auth_config.auth_methods.get(auth_config.settings.auth_method)
|
auth = auth_config.auth_methods.get(auth_config.settings.auth_method)
|
||||||
return auth
|
if auth:
|
||||||
|
return auth()
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -7,7 +7,7 @@ config.settings = config.Settings(_env_file='tests/test.env')
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
from httpx2 import ASGITransport, AsyncClient
|
from httpx2 import ASGITransport, AsyncClient, BasicAuth
|
||||||
|
|
||||||
from scopes.api import main
|
from scopes.api import main
|
||||||
from scopes.api.auth import config as auth_config
|
from scopes.api.auth import config as auth_config
|
||||||
|
|
@ -15,6 +15,9 @@ from scopes.api.main import app
|
||||||
from scopes.storage import tracking, user as st_user
|
from scopes.storage import tracking, user as st_user
|
||||||
from scopes.api.auth import user as api_user
|
from scopes.api.auth import user as api_user
|
||||||
|
|
||||||
|
config.settings = config.Settings(_env_file='tests/test.env')
|
||||||
|
auth_config.settings = auth_config.Settings(_env_file='tests/test.env')
|
||||||
|
|
||||||
@pytest.fixture(scope='module')
|
@pytest.fixture(scope='module')
|
||||||
def storage():
|
def storage():
|
||||||
storage = config.StorageFactory(config.settings)(config.settings.dbschema)
|
storage = config.StorageFactory(config.settings)(config.settings.dbschema)
|
||||||
|
|
@ -73,13 +76,14 @@ def test_user(storage, caplog):
|
||||||
storage.commit()
|
storage.commit()
|
||||||
with TestClient(app) as clt:
|
with TestClient(app) as clt:
|
||||||
caplog.set_level(config.settings.log_level)
|
caplog.set_level(config.settings.log_level)
|
||||||
|
auth = BasicAuth(username='jim', password='secret')
|
||||||
users = clt.get('/users').json()
|
users = clt.get('/users').json()
|
||||||
assert len(users) == 1
|
assert len(users) == 1
|
||||||
u01i = users[0]
|
u01i = users[0]
|
||||||
assert u01i.get(pw_hash) is None
|
assert u01i.get(pw_hash) is None
|
||||||
inp = {'name': 'jane', 'login': 'jane', 'display_name': 'Jane',
|
inp = {'name': 'jane', 'login': 'jane', 'display_name': 'Jane',
|
||||||
'email': 'jane@xmpl.org', 'password': 'secret'}
|
'email': 'jane@xmpl.org', 'password': 'secret'}
|
||||||
resp = clt.post('/users', json=inp)
|
resp = clt.post('/users', json=inp, auth=auth)
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
u02 = resp.json()
|
u02 = resp.json()
|
||||||
assert u02['uid'] == 'user-2'
|
assert u02['uid'] == 'user-2'
|
||||||
|
|
@ -87,7 +91,7 @@ def test_user(storage, caplog):
|
||||||
assert ts02 is not None
|
assert ts02 is not None
|
||||||
del inp['password']
|
del inp['password']
|
||||||
inp['display_name'] = 'Jane Doe'
|
inp['display_name'] = 'Jane Doe'
|
||||||
resp = clt.post('/users', json=inp)
|
resp = clt.post('/users', json=inp, auth=auth)
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
u02a = resp.json()
|
u02a = resp.json()
|
||||||
assert u02a['uid'] == 'user-2'
|
assert u02a['uid'] == 'user-2'
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue