Changeset View
Changeset View
Standalone View
Standalone View
swh/web/client/tests/test_web_api_client.py
| # Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| from copy import copy | |||||
| from datetime import datetime | |||||
| from dateutil.parser import parse as parse_date | from dateutil.parser import parse as parse_date | ||||
| from unittest.mock import call, Mock | |||||
| import pytest | |||||
| from swh.web.client.auth import AuthenticationError | |||||
| from swh.model.identifiers import parse_persistent_identifier as parse_pid | from swh.model.identifiers import parse_persistent_identifier as parse_pid | ||||
| from .test_cli import oidc_profile | |||||
| def test_get_content(web_api_client, web_api_mock): | def test_get_content(web_api_client, web_api_mock): | ||||
| pid = parse_pid("swh:1:cnt:fe95a46679d128ff167b7c55df5d02356c5a1ae1") | pid = parse_pid("swh:1:cnt:fe95a46679d128ff167b7c55df5d02356c5a1ae1") | ||||
| obj = web_api_client.get(pid) | obj = web_api_client.get(pid) | ||||
| assert obj["length"] == 151810 | assert obj["length"] == 151810 | ||||
| for key in ("length", "status", "checksums", "data_url"): | for key in ("length", "status", "checksums", "data_url"): | ||||
| assert key in obj | assert key in obj | ||||
| ▲ Show 20 Lines • Show All 79 Lines • ▼ Show 20 Lines | def test_iter_snapshot(web_api_client, web_api_mock): | ||||
| pid = parse_pid("swh:1:snp:cabcc7d7bf639bbe1cc3b41989e1806618dd5764") | pid = parse_pid("swh:1:snp:cabcc7d7bf639bbe1cc3b41989e1806618dd5764") | ||||
| obj = web_api_client.snapshot(pid) | obj = web_api_client.snapshot(pid) | ||||
| snp = {} | snp = {} | ||||
| for partial in obj: | for partial in obj: | ||||
| snp.update(partial) | snp.update(partial) | ||||
| assert len(snp) == 1391 | assert len(snp) == 1391 | ||||
| def test_authenticate_success(web_api_client, web_api_mock): | |||||
| rel_id = 'b9db10d00835e9a43e2eebef2db1d04d4ae82342' | |||||
| url = f'{web_api_client.api_url}/release/{rel_id}/' | |||||
| web_api_client.oidc_session = Mock() | |||||
| web_api_client.oidc_session.refresh.return_value = copy(oidc_profile) | |||||
| access_token = oidc_profile['access_token'] | |||||
| refresh_token = 'user-refresh-token' | |||||
| web_api_client.authenticate(refresh_token) | |||||
| assert 'expires_at' in web_api_client.oidc_profile | |||||
| pid = parse_pid(f'swh:1:rel:{rel_id}') | |||||
| web_api_client.get(pid) | |||||
| web_api_client.oidc_session.refresh.assert_called_once_with(refresh_token) | |||||
| sent_request = web_api_mock._adapter.last_request | |||||
| assert sent_request.url == url | |||||
| assert 'Authorization' in sent_request.headers | |||||
| assert sent_request.headers['Authorization'] == f'Bearer {access_token}' | |||||
| def test_authenticate_refresh_token(web_api_client, web_api_mock): | |||||
| rel_id = 'b9db10d00835e9a43e2eebef2db1d04d4ae82342' | |||||
| url = f'{web_api_client.api_url}/release/{rel_id}/' | |||||
| oidc_profile_cp = copy(oidc_profile) | |||||
| web_api_client.oidc_session = Mock() | |||||
| web_api_client.oidc_session.refresh.return_value = oidc_profile_cp | |||||
| refresh_token = 'user-refresh-token' | |||||
| web_api_client.authenticate(refresh_token) | |||||
| assert 'expires_at' in web_api_client.oidc_profile | |||||
| # simulate access token expiration | |||||
| web_api_client.oidc_profile['expires_at'] = datetime.now() | |||||
| access_token = 'new-access-token' | |||||
| oidc_profile_cp['access_token'] = access_token | |||||
| pid = parse_pid(f'swh:1:rel:{rel_id}') | |||||
| web_api_client.get(pid) | |||||
| calls = [call(refresh_token), call(oidc_profile['refresh_token'])] | |||||
| web_api_client.oidc_session.refresh.assert_has_calls(calls) | |||||
| sent_request = web_api_mock._adapter.last_request | |||||
| assert sent_request.url == url | |||||
| assert 'Authorization' in sent_request.headers | |||||
| assert sent_request.headers['Authorization'] == f'Bearer {access_token}' | |||||
| def test_authenticate_failure(web_api_client, web_api_mock): | |||||
| msg = 'Authentication error' | |||||
| web_api_client.oidc_session = Mock() | |||||
| web_api_client.oidc_session.refresh.side_effect = Exception(msg) | |||||
| refresh_token = 'user-refresh-token' | |||||
| with pytest.raises(AuthenticationError) as e: | |||||
| web_api_client.authenticate(refresh_token) | |||||
| assert e.match(msg) | |||||
| oidc_error_response = { | |||||
| 'error': 'invalid_grant', | |||||
| 'error_description': 'Invalid refresh token', | |||||
| } | |||||
| web_api_client.oidc_session.refresh.side_effect = None | |||||
| web_api_client.oidc_session.refresh.return_value = oidc_error_response | |||||
| with pytest.raises(AuthenticationError) as e: | |||||
| web_api_client.authenticate(refresh_token) | |||||
| assert e.match(repr(oidc_error_response)) | |||||