forked from platypush/platypush
Fixed tests
This commit is contained in:
parent
958ef6b987
commit
6ec8a991df
3 changed files with 48 additions and 18 deletions
|
@ -191,8 +191,9 @@ def authenticate(
|
||||||
check_csrf_token=False,
|
check_csrf_token=False,
|
||||||
json=False,
|
json=False,
|
||||||
):
|
):
|
||||||
def on_auth_fail():
|
def on_auth_fail(has_users=True):
|
||||||
if json:
|
if json:
|
||||||
|
if has_users:
|
||||||
return (
|
return (
|
||||||
jsonify(
|
jsonify(
|
||||||
{
|
{
|
||||||
|
@ -202,6 +203,16 @@ def authenticate(
|
||||||
401,
|
401,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
return (
|
||||||
|
jsonify(
|
||||||
|
{
|
||||||
|
'message': 'Please register a user through '
|
||||||
|
'the web panel first',
|
||||||
|
}
|
||||||
|
),
|
||||||
|
412,
|
||||||
|
)
|
||||||
|
|
||||||
return redirect('/login?redirect=' + (redirect_page or request.url), 307)
|
return redirect('/login?redirect=' + (redirect_page or request.url), 307)
|
||||||
|
|
||||||
def decorator(f):
|
def decorator(f):
|
||||||
|
@ -241,7 +252,7 @@ def authenticate(
|
||||||
return abort(403, 'Invalid or missing csrf_token')
|
return abort(403, 'Invalid or missing csrf_token')
|
||||||
|
|
||||||
if n_users == 0 and 'session' not in skip_methods:
|
if n_users == 0 and 'session' not in skip_methods:
|
||||||
return on_auth_fail()
|
return on_auth_fail(has_users=False)
|
||||||
|
|
||||||
if (
|
if (
|
||||||
('http' not in skip_methods and http_auth_ok)
|
('http' not in skip_methods and http_auth_ok)
|
||||||
|
|
|
@ -1,5 +1,6 @@
|
||||||
import inspect
|
import inspect
|
||||||
import pathlib
|
import pathlib
|
||||||
|
import types
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Mapping, Type, Tuple, Any
|
from typing import Mapping, Type, Tuple, Any
|
||||||
|
|
||||||
|
@ -49,7 +50,10 @@ class Entity(Base):
|
||||||
|
|
||||||
UniqueConstraint(external_id, plugin)
|
UniqueConstraint(external_id, plugin)
|
||||||
|
|
||||||
__table_args__ = (Index(name, plugin), Index(name, type, plugin))
|
__table_args__ = (
|
||||||
|
Index('name_and_plugin_index', name, plugin),
|
||||||
|
Index('name_type_and_plugin_index', name, type, plugin),
|
||||||
|
)
|
||||||
|
|
||||||
__mapper_args__ = {
|
__mapper_args__ = {
|
||||||
'polymorphic_identity': __tablename__,
|
'polymorphic_identity': __tablename__,
|
||||||
|
@ -104,9 +108,10 @@ def _discover_entity_types():
|
||||||
onerror=lambda _: None,
|
onerror=lambda _: None,
|
||||||
):
|
):
|
||||||
try:
|
try:
|
||||||
mod_loader = loader.find_module(modname) # type: ignore
|
mod_loader = loader.find_spec(modname, None)
|
||||||
assert mod_loader
|
assert mod_loader and mod_loader.loader
|
||||||
module = mod_loader.load_module() # type: ignore
|
module = types.ModuleType(mod_loader.name)
|
||||||
|
mod_loader.loader.exec_module(module)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f'Could not import module {modname}')
|
logger.warning(f'Could not import module {modname}')
|
||||||
logger.exception(e)
|
logger.exception(e)
|
||||||
|
|
|
@ -22,8 +22,10 @@ def test_request_with_no_registered_users(base_url, expected_registration_redire
|
||||||
An /execute request performed before any user is registered should redirect to the registration page.
|
An /execute request performed before any user is registered should redirect to the registration page.
|
||||||
"""
|
"""
|
||||||
response = send_request(authenticate=False, parse_json=False)
|
response = send_request(authenticate=False, parse_json=False)
|
||||||
assert expected_registration_redirect == response.url, \
|
assert response.status_code == 412, (
|
||||||
'No users registered, but the application did not redirect us to the registration page'
|
'No users registered, but the execute endpoint returned '
|
||||||
|
f'{response.status_code}'
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_first_user_registration(base_url):
|
def test_first_user_registration(base_url):
|
||||||
|
@ -33,9 +35,12 @@ def test_first_user_registration(base_url):
|
||||||
response = register_user()
|
response = register_user()
|
||||||
|
|
||||||
assert len(response.history) > 0, 'Redirect missing from the history'
|
assert len(response.history) > 0, 'Redirect missing from the history'
|
||||||
assert 'session_token' in response.history[0].cookies, 'No session_token returned upon registration'
|
assert (
|
||||||
assert '{base_url}/'.format(base_url=base_url) == response.url, \
|
'session_token' in response.history[0].cookies
|
||||||
'The registration form did not redirect to the main panel'
|
), 'No session_token returned upon registration'
|
||||||
|
assert (
|
||||||
|
'{base_url}/'.format(base_url=base_url) == response.url
|
||||||
|
), 'The registration form did not redirect to the main panel'
|
||||||
|
|
||||||
|
|
||||||
def test_unauthorized_request_with_registered_user(base_url, expected_login_redirect):
|
def test_unauthorized_request_with_registered_user(base_url, expected_login_redirect):
|
||||||
|
@ -43,20 +48,29 @@ def test_unauthorized_request_with_registered_user(base_url, expected_login_redi
|
||||||
After a first user has been registered any unauthenticated call to /execute should redirect to /login.
|
After a first user has been registered any unauthenticated call to /execute should redirect to /login.
|
||||||
"""
|
"""
|
||||||
response = send_request(authenticate=False, parse_json=False)
|
response = send_request(authenticate=False, parse_json=False)
|
||||||
assert expected_login_redirect == response.url, \
|
assert response.status_code == 401, (
|
||||||
'An unauthenticated request after user registration should result in a login redirect'
|
'An unauthenticated request after user registration should result in a '
|
||||||
|
f'401 error, got {response.status_code} instead'
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_authorized_request_with_registered_user(base_url):
|
def test_authorized_request_with_registered_user(base_url):
|
||||||
# A request authenticated with user/pass should succeed.
|
# A request authenticated with user/pass should succeed.
|
||||||
response = send_request(authenticate=True)
|
response = send_request(authenticate=True)
|
||||||
assert response.output.strip() == 'ping', 'The request did not return the expected output'
|
assert (
|
||||||
|
response.output.strip() == 'ping'
|
||||||
|
), 'The request did not return the expected output'
|
||||||
|
|
||||||
|
|
||||||
def test_request_with_wrong_credentials(base_url, expected_login_redirect):
|
def test_request_with_wrong_credentials(base_url, expected_login_redirect):
|
||||||
# A request with the wrong user/pass should fail.
|
# A request with the wrong user/pass should fail.
|
||||||
response = send_request(authenticate=False, auth=('wrong', 'wrong'), parse_json=False)
|
response = send_request(
|
||||||
assert expected_login_redirect == response.url, 'A request with wrong credentials should fail'
|
authenticate=False, auth=('wrong', 'wrong'), parse_json=False
|
||||||
|
)
|
||||||
|
assert response.status_code == 401, (
|
||||||
|
'A request with wrong credentials should fail with status 401, '
|
||||||
|
f'got {response.status_code} instead'
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
|
|
Loading…
Reference in a new issue