forked from platypush/platypush
[tests] Fixed tests after HTTP auth refactor.
This commit is contained in:
parent
70db33b4e2
commit
b1b51b4b7e
3 changed files with 17 additions and 18 deletions
|
@ -40,11 +40,13 @@ def _wait_for_app(app: Application, timeout: int = app_start_timeout):
|
||||||
|
|
||||||
while not success and time.time() - start_time < timeout:
|
while not success and time.time() - start_time < timeout:
|
||||||
try:
|
try:
|
||||||
response = requests.get(f'http://localhost:{http.port}/')
|
response = requests.get(
|
||||||
|
f'http://localhost:{http.port}/', timeout=1, allow_redirects=False
|
||||||
|
)
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
success = True
|
success = True
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.debug('App not ready yet: %s', e)
|
logging.info('App not ready yet: %s', e)
|
||||||
time.sleep(1)
|
time.sleep(1)
|
||||||
|
|
||||||
assert success, f'App not ready after {timeout} seconds'
|
assert success, f'App not ready after {timeout} seconds'
|
||||||
|
|
|
@ -5,12 +5,12 @@ from .utils import register_user, send_request as _send_request
|
||||||
|
|
||||||
@pytest.fixture(scope='module')
|
@pytest.fixture(scope='module')
|
||||||
def expected_registration_redirect(base_url):
|
def expected_registration_redirect(base_url):
|
||||||
yield '{base_url}/register?redirect={base_url}/execute'.format(base_url=base_url)
|
yield f'{base_url}/auth?type=register&redirect={base_url}/execute'
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope='module')
|
@pytest.fixture(scope='module')
|
||||||
def expected_login_redirect(base_url):
|
def expected_login_redirect(base_url):
|
||||||
yield '{base_url}/login?redirect={base_url}/execute'.format(base_url=base_url)
|
yield f'{base_url}/auth?type=login&redirect={base_url}/execute'
|
||||||
|
|
||||||
|
|
||||||
def send_request(**kwargs):
|
def send_request(**kwargs):
|
||||||
|
@ -34,18 +34,14 @@ def test_first_user_registration(base_url):
|
||||||
"""
|
"""
|
||||||
response = register_user()
|
response = register_user()
|
||||||
|
|
||||||
assert len(response.history) > 0, 'Redirect missing from the history'
|
assert response.json().get('status') == 'ok' and response.json().get(
|
||||||
assert (
|
'session_token'
|
||||||
'session_token' in response.history[0].cookies
|
|
||||||
), 'No session_token returned upon registration'
|
), '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):
|
||||||
"""
|
"""
|
||||||
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 /auth.
|
||||||
"""
|
"""
|
||||||
response = send_request(authenticate=False, parse_json=False)
|
response = send_request(authenticate=False, parse_json=False)
|
||||||
assert response.status_code == 401, (
|
assert response.status_code == 401, (
|
||||||
|
|
|
@ -48,7 +48,7 @@ def send_request(
|
||||||
args: Optional[dict] = None,
|
args: Optional[dict] = None,
|
||||||
parse_json: bool = True,
|
parse_json: bool = True,
|
||||||
authenticate: bool = True,
|
authenticate: bool = True,
|
||||||
**kwargs
|
**kwargs,
|
||||||
):
|
):
|
||||||
if not timeout:
|
if not timeout:
|
||||||
timeout = request_timeout
|
timeout = request_timeout
|
||||||
|
@ -58,14 +58,15 @@ def send_request(
|
||||||
auth = (test_user, test_pass) if authenticate else kwargs.pop('auth', ())
|
auth = (test_user, test_pass) if authenticate else kwargs.pop('auth', ())
|
||||||
set_timeout(seconds=timeout, on_timeout=on_timeout('Receiver response timed out'))
|
set_timeout(seconds=timeout, on_timeout=on_timeout('Receiver response timed out'))
|
||||||
response = requests.post(
|
response = requests.post(
|
||||||
'{}/execute'.format(base_url),
|
f'{base_url}/execute',
|
||||||
auth=auth,
|
auth=auth,
|
||||||
|
timeout=10,
|
||||||
json={
|
json={
|
||||||
'type': 'request',
|
'type': 'request',
|
||||||
'action': action,
|
'action': action,
|
||||||
'args': args,
|
'args': args,
|
||||||
},
|
},
|
||||||
**kwargs
|
**kwargs,
|
||||||
)
|
)
|
||||||
|
|
||||||
clear_timeout()
|
clear_timeout()
|
||||||
|
@ -85,7 +86,7 @@ def register_user(username: Optional[str] = None, password: Optional[str] = None
|
||||||
on_timeout=on_timeout('User registration response timed out'),
|
on_timeout=on_timeout('User registration response timed out'),
|
||||||
)
|
)
|
||||||
response = requests.post(
|
response = requests.post(
|
||||||
'{base_url}/register?redirect={base_url}/'.format(base_url=base_url),
|
f'{base_url}/auth?type=register&redirect={base_url}/',
|
||||||
data={
|
data={
|
||||||
'username': username,
|
'username': username,
|
||||||
'password': password,
|
'password': password,
|
||||||
|
@ -106,7 +107,7 @@ def on_timeout(msg):
|
||||||
|
|
||||||
def parse_response(response):
|
def parse_response(response):
|
||||||
response = Message.build(response.json())
|
response = Message.build(response.json())
|
||||||
assert isinstance(response, Response), 'Expected Response type, got {}'.format(
|
assert isinstance(
|
||||||
response.__class__.__name__
|
response, Response
|
||||||
)
|
), f'Expected Response type, got {response.__class__.__name__}'
|
||||||
return response
|
return response
|
||||||
|
|
Loading…
Reference in a new issue