mirror of
https://github.com/jupyterhub/jupyterhub.git
synced 2025-10-15 05:53:00 +00:00
resolved conflicts with rbac branch
This commit is contained in:
@@ -14,11 +14,9 @@ from pytest import mark
|
||||
|
||||
import jupyterhub
|
||||
from .. import orm
|
||||
from .. import roles
|
||||
from ..objects import Server
|
||||
from ..utils import url_path_join as ujoin
|
||||
from ..utils import utcnow
|
||||
from .mocking import mock_role
|
||||
from .mocking import public_host
|
||||
from .mocking import public_url
|
||||
from .utils import add_user
|
||||
@@ -27,6 +25,7 @@ from .utils import async_requests
|
||||
from .utils import auth_header
|
||||
from .utils import find_user
|
||||
|
||||
|
||||
# --------------------
|
||||
# Authentication tests
|
||||
# --------------------
|
||||
@@ -175,15 +174,20 @@ async def test_get_users(app):
|
||||
|
||||
users = sorted(r.json(), key=lambda d: d['name'])
|
||||
users = [normalize_user(u) for u in users]
|
||||
user_model = {
|
||||
'name': 'user',
|
||||
'admin': False,
|
||||
'roles': ['user'],
|
||||
'last_activity': None,
|
||||
}
|
||||
assert users == [
|
||||
fill_user({'name': 'admin', 'admin': True, 'roles': ['admin']}),
|
||||
fill_user(
|
||||
{'name': 'user', 'admin': False, 'roles': ['user'], 'last_activity': None}
|
||||
),
|
||||
fill_user(user_model),
|
||||
]
|
||||
with mock_role(app, 'user'):
|
||||
r = await api_request(app, 'users', headers=auth_header(db, 'user'))
|
||||
assert r.status_code == 403
|
||||
r = await api_request(app, 'users', headers=auth_header(db, 'user'))
|
||||
assert r.status_code == 200
|
||||
r_user_model = r.json()[0]
|
||||
assert r_user_model['name'] == user_model['name']
|
||||
|
||||
|
||||
@mark.user
|
||||
@@ -284,15 +288,25 @@ async def test_get_self(app):
|
||||
assert model['name'] == u.name
|
||||
|
||||
# invalid auth gets 403
|
||||
with mock_role(app, 'user'):
|
||||
r = await api_request(
|
||||
app,
|
||||
'user',
|
||||
headers={'Authorization': 'token notvalid'},
|
||||
)
|
||||
r = await api_request(
|
||||
app,
|
||||
'user',
|
||||
headers={'Authorization': 'token notvalid'},
|
||||
)
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
async def test_get_self_service(app, mockservice):
|
||||
r = await api_request(
|
||||
app, "user", headers={"Authorization": f"token {mockservice.api_token}"}
|
||||
)
|
||||
r.raise_for_status()
|
||||
service_info = r.json()
|
||||
|
||||
assert service_info['kind'] == 'service'
|
||||
assert service_info['name'] == mockservice.name
|
||||
|
||||
|
||||
@mark.user
|
||||
@mark.role
|
||||
async def test_add_user(app):
|
||||
@@ -314,12 +328,11 @@ async def test_add_user(app):
|
||||
async def test_get_user(app):
|
||||
name = 'user'
|
||||
_ = await api_request(app, 'users', name, headers=auth_header(app.db, name))
|
||||
with mock_role(app, role=name, name=name):
|
||||
r = await api_request(
|
||||
app,
|
||||
'users',
|
||||
name,
|
||||
)
|
||||
r = await api_request(
|
||||
app,
|
||||
'users',
|
||||
name,
|
||||
)
|
||||
assert r.status_code == 200
|
||||
|
||||
user = normalize_user(r.json())
|
||||
@@ -518,15 +531,14 @@ async def test_user_set_auth_state(app, auth_state_enabled):
|
||||
assert user.name == name
|
||||
user_auth_state = await user.get_auth_state()
|
||||
assert user_auth_state is None
|
||||
with mock_role(app, 'user'):
|
||||
r = await api_request(
|
||||
app,
|
||||
'users',
|
||||
name,
|
||||
method='patch',
|
||||
data=json.dumps({'auth_state': auth_state}),
|
||||
headers=auth_header(app.db, name),
|
||||
)
|
||||
r = await api_request(
|
||||
app,
|
||||
'users',
|
||||
name,
|
||||
method='patch',
|
||||
data=json.dumps({'auth_state': auth_state}),
|
||||
headers=auth_header(app.db, name),
|
||||
)
|
||||
assert r.status_code == 403
|
||||
user_auth_state = await user.get_auth_state()
|
||||
assert user_auth_state is None
|
||||
@@ -1348,16 +1360,15 @@ async def test_token_authenticator_noauth(app):
|
||||
"""Create a token for a user relying on Authenticator.authenticate and no auth header"""
|
||||
name = 'user'
|
||||
data = {'auth': {'username': name, 'password': name}}
|
||||
with mock_role(app, 'admin'):
|
||||
r = await api_request(
|
||||
app,
|
||||
'users',
|
||||
name,
|
||||
'tokens',
|
||||
method='post',
|
||||
data=json.dumps(data) if data else None,
|
||||
noauth=True,
|
||||
)
|
||||
r = await api_request(
|
||||
app,
|
||||
'users',
|
||||
name,
|
||||
'tokens',
|
||||
method='post',
|
||||
data=json.dumps(data) if data else None,
|
||||
noauth=True,
|
||||
)
|
||||
assert r.status_code == 200
|
||||
reply = r.json()
|
||||
assert 'token' in reply
|
||||
@@ -1372,16 +1383,15 @@ async def test_token_authenticator_dict_noauth(app):
|
||||
app.authenticator.auth_state = {'who': 'cares'}
|
||||
name = 'user'
|
||||
data = {'auth': {'username': name, 'password': name}}
|
||||
with mock_role(app, 'user'):
|
||||
r = await api_request(
|
||||
app,
|
||||
'users',
|
||||
name,
|
||||
'tokens',
|
||||
method='post',
|
||||
data=json.dumps(data) if data else None,
|
||||
noauth=True,
|
||||
)
|
||||
r = await api_request(
|
||||
app,
|
||||
'users',
|
||||
name,
|
||||
'tokens',
|
||||
method='post',
|
||||
data=json.dumps(data) if data else None,
|
||||
noauth=True,
|
||||
)
|
||||
assert r.status_code == 200
|
||||
reply = r.json()
|
||||
assert 'token' in reply
|
||||
@@ -1396,7 +1406,7 @@ async def test_token_authenticator_dict_noauth(app):
|
||||
[
|
||||
('admin', 'other', 200),
|
||||
('admin', 'missing', 404),
|
||||
('user', 'other', 403),
|
||||
('user', 'other', 404),
|
||||
('user', 'user', 200),
|
||||
],
|
||||
)
|
||||
@@ -1405,8 +1415,7 @@ async def test_token_list(app, as_user, for_user, status):
|
||||
if for_user != 'missing':
|
||||
for_user_obj = add_user(app.db, app, name=for_user)
|
||||
headers = {'Authorization': 'token %s' % u.new_api_token()}
|
||||
with mock_role(app, role=as_user, name=as_user):
|
||||
r = await api_request(app, 'users', for_user, 'tokens', headers=headers)
|
||||
r = await api_request(app, 'users', for_user, 'tokens', headers=headers)
|
||||
assert r.status_code == status
|
||||
if status != 200:
|
||||
return
|
||||
@@ -1417,10 +1426,9 @@ async def test_token_list(app, as_user, for_user, status):
|
||||
assert all(token['user'] == for_user for token in reply['oauth_tokens'])
|
||||
# validate individual token ids
|
||||
for token in reply['api_tokens'] + reply['oauth_tokens']:
|
||||
with mock_role(app, role=as_user, name=as_user):
|
||||
r = await api_request(
|
||||
app, 'users', for_user, 'tokens', token['id'], headers=headers
|
||||
)
|
||||
r = await api_request(
|
||||
app, 'users', for_user, 'tokens', token['id'], headers=headers
|
||||
)
|
||||
r.raise_for_status()
|
||||
reply = r.json()
|
||||
assert normalize_token(reply) == normalize_token(token)
|
||||
@@ -1608,8 +1616,7 @@ async def test_get_services(app, mockservice_url):
|
||||
'display': True,
|
||||
}
|
||||
}
|
||||
with mock_role(app, 'user'):
|
||||
r = await api_request(app, 'services', headers=auth_header(db, 'user'))
|
||||
r = await api_request(app, 'services', headers=auth_header(db, 'user'))
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
@@ -1633,17 +1640,15 @@ async def test_get_service(app, mockservice_url):
|
||||
'info': {},
|
||||
'display': True,
|
||||
}
|
||||
with mock_role(app, 'service'):
|
||||
r = await api_request(
|
||||
app,
|
||||
'services/%s' % mockservice.name,
|
||||
headers={'Authorization': 'token %s' % mockservice.api_token},
|
||||
)
|
||||
r.raise_for_status()
|
||||
with mock_role(app, 'user'):
|
||||
r = await api_request(
|
||||
app, 'services/%s' % mockservice.name, headers=auth_header(db, 'user')
|
||||
)
|
||||
r = await api_request(
|
||||
app,
|
||||
'services/%s' % mockservice.name,
|
||||
headers={'Authorization': 'token %s' % mockservice.api_token},
|
||||
)
|
||||
r.raise_for_status()
|
||||
r = await api_request(
|
||||
app, 'services/%s' % mockservice.name, headers=auth_header(db, 'user')
|
||||
)
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
@@ -1654,7 +1659,7 @@ async def test_root_api(app):
|
||||
if app.internal_ssl:
|
||||
kwargs['cert'] = (app.internal_ssl_cert, app.internal_ssl_key)
|
||||
kwargs["verify"] = app.internal_ssl_ca
|
||||
r = await async_requests.get(url, **kwargs)
|
||||
r = await api_request(app, bypass_proxy=True)
|
||||
r.raise_for_status()
|
||||
expected = {'version': jupyterhub.__version__}
|
||||
assert r.json() == expected
|
||||
@@ -1691,15 +1696,14 @@ async def test_info(app):
|
||||
|
||||
async def test_update_activity_403(app, user, admin_user):
|
||||
token = user.new_api_token()
|
||||
with mock_role(app, 'user'):
|
||||
r = await api_request(
|
||||
app,
|
||||
"users/{}/activity".format(admin_user.name),
|
||||
headers={"Authorization": "token {}".format(token)},
|
||||
data="{}",
|
||||
method="post",
|
||||
)
|
||||
assert r.status_code == 403
|
||||
r = await api_request(
|
||||
app,
|
||||
"users/{}/activity".format(admin_user.name),
|
||||
headers={"Authorization": "token {}".format(token)},
|
||||
data="{}",
|
||||
method="post",
|
||||
)
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
async def test_update_activity_admin(app, user, admin_user):
|
||||
|
Reference in New Issue
Block a user