Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/api/views/test_origin.py
| # Copyright (C) 2015-2018 The Software Heritage developers | # Copyright (C) 2015-2018 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 Affero General Public License version 3, or any later version | # License: GNU Affero 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 | ||||
| import random | import random | ||||
| from hypothesis import given | from hypothesis import given | ||||
| from hypothesis.strategies import integers | |||||
| from rest_framework.test import APITestCase | from rest_framework.test import APITestCase | ||||
| from unittest.mock import patch | from unittest.mock import patch | ||||
| from swh.storage.exc import StorageDBError, StorageAPIError | from swh.storage.exc import StorageDBError, StorageAPIError | ||||
| from swh.model.hashutil import hash_to_hex | |||||
| from swh.web.common.utils import reverse | from swh.web.common.utils import reverse | ||||
| from swh.web.common.origin_visits import get_origin_visits | from swh.web.common.origin_visits import get_origin_visits | ||||
| from swh.web.tests.strategies import ( | from swh.web.tests.strategies import ( | ||||
| origin, new_origin, new_origins, visit_dates, new_snapshots | origin, new_origin, new_origins, visit_dates, new_snapshots | ||||
| ) | ) | ||||
| from swh.web.tests.testcase import WebTestCase | from swh.web.tests.testcase import WebTestCase | ||||
| class OriginApiTestCase(WebTestCase, APITestCase): | class OriginApiTestCase(WebTestCase, APITestCase): | ||||
| @patch('swh.web.api.views.origin.get_origin_visits') | @patch('swh.web.api.views.origin.get_origin_visits') | ||||
| def test_api_lookup_origin_visits_raise_error( | def test_api_lookup_origin_visits_raise_error( | ||||
| self, mock_get_origin_visits, | self, mock_get_origin_visits, | ||||
| ): | ): | ||||
| err_msg = 'voluntary error to check the bad request middleware.' | err_msg = 'voluntary error to check the bad request middleware.' | ||||
| mock_get_origin_visits.side_effect = ValueError(err_msg) | mock_get_origin_visits.side_effect = ValueError(err_msg) | ||||
| url = reverse('api-origin-visits', url_args={'origin_id': 2}) | url = reverse('api-origin-visits', url_args={'origin_id': '42'*20}) | ||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| self.assertEqual(rv.status_code, 400) | self.assertEqual(rv.status_code, 400) | ||||
| self.assertEqual(rv['Content-Type'], 'application/json') | self.assertEqual(rv['Content-Type'], 'application/json') | ||||
| self.assertEqual(rv.data, { | self.assertEqual(rv.data, { | ||||
| 'exception': 'ValueError', | 'exception': 'ValueError', | ||||
| 'reason': err_msg}) | 'reason': err_msg}) | ||||
| @patch('swh.web.api.views.origin.get_origin_visits') | @patch('swh.web.api.views.origin.get_origin_visits') | ||||
| def test_api_lookup_origin_visits_raise_swh_storage_error_db( | def test_api_lookup_origin_visits_raise_swh_storage_error_db( | ||||
| self, mock_get_origin_visits): | self, mock_get_origin_visits): | ||||
| err_msg = 'Storage exploded! Will be back online shortly!' | err_msg = 'Storage exploded! Will be back online shortly!' | ||||
| mock_get_origin_visits.side_effect = StorageDBError(err_msg) | mock_get_origin_visits.side_effect = StorageDBError(err_msg) | ||||
| url = reverse('api-origin-visits', url_args={'origin_id': 2}) | url = reverse('api-origin-visits', url_args={'origin_id': '42'*20}) | ||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| self.assertEqual(rv.status_code, 503) | self.assertEqual(rv.status_code, 503, rv.data) | ||||
| self.assertEqual(rv['Content-Type'], 'application/json') | self.assertEqual(rv['Content-Type'], 'application/json') | ||||
| self.assertEqual(rv.data, { | self.assertEqual(rv.data, { | ||||
| 'exception': 'StorageDBError', | 'exception': 'StorageDBError', | ||||
| 'reason': | 'reason': | ||||
| 'An unexpected error occurred in the backend: %s' % err_msg}) | 'An unexpected error occurred in the backend: %s' % err_msg}) | ||||
| @patch('swh.web.api.views.origin.get_origin_visits') | @patch('swh.web.api.views.origin.get_origin_visits') | ||||
| def test_api_lookup_origin_visits_raise_swh_storage_error_api( | def test_api_lookup_origin_visits_raise_swh_storage_error_api( | ||||
| self, mock_get_origin_visits): | self, mock_get_origin_visits): | ||||
| err_msg = 'Storage API dropped dead! Will resurrect asap!' | err_msg = 'Storage API dropped dead! Will resurrect asap!' | ||||
| mock_get_origin_visits.side_effect = StorageAPIError(err_msg) | mock_get_origin_visits.side_effect = StorageAPIError(err_msg) | ||||
| url = reverse('api-origin-visits', url_args={'origin_id': 2}) | url = reverse('api-origin-visits', url_args={'origin_id': '42'*20}) | ||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| self.assertEqual(rv.status_code, 503) | self.assertEqual(rv.status_code, 503) | ||||
| self.assertEqual(rv['Content-Type'], 'application/json') | self.assertEqual(rv['Content-Type'], 'application/json') | ||||
| self.assertEqual(rv.data, { | self.assertEqual(rv.data, { | ||||
| 'exception': 'StorageAPIError', | 'exception': 'StorageAPIError', | ||||
| 'reason': | 'reason': | ||||
| 'An unexpected error occurred in the api backend: %s' % err_msg | 'An unexpected error occurred in the api backend: %s' % err_msg | ||||
| Show All 12 Lines | def test_api_lookup_origin_visits(self, new_origin, visit_dates, | ||||
| all_visits = list(reversed(get_origin_visits(new_origin))) | all_visits = list(reversed(get_origin_visits(new_origin))) | ||||
| for last_visit, expected_visits in ( | for last_visit, expected_visits in ( | ||||
| (None, all_visits[:2]), | (None, all_visits[:2]), | ||||
| (all_visits[1]['visit'], all_visits[2:4])): | (all_visits[1]['visit'], all_visits[2:4])): | ||||
| url = reverse('api-origin-visits', | url = reverse('api-origin-visits', | ||||
| url_args={'origin_id': origin_id}, | url_args={'origin_id': hash_to_hex(origin_id)}, | ||||
| query_params={'per_page': 2, | query_params={'per_page': 2, | ||||
| 'last_visit': last_visit}) | 'last_visit': last_visit}) | ||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| self.assertEqual(rv.status_code, 200) | self.assertEqual(rv.status_code, 200, rv.data) | ||||
| self.assertEqual(rv['Content-Type'], 'application/json') | self.assertEqual(rv['Content-Type'], 'application/json') | ||||
| for expected_visit in expected_visits: | for expected_visit in expected_visits: | ||||
| origin_visit_url = reverse( | origin_visit_url = reverse( | ||||
| 'api-origin-visit', | 'api-origin-visit', | ||||
| url_args={'origin_id': origin_id, | url_args={'origin_id': hash_to_hex(origin_id), | ||||
| 'visit_id': expected_visit['visit']}) | 'visit_id': expected_visit['visit']}) | ||||
| snapshot_url = reverse( | snapshot_url = reverse( | ||||
| 'api-snapshot', | 'api-snapshot', | ||||
| url_args={'snapshot_id': expected_visit['snapshot']}) | url_args={'snapshot_id': expected_visit['snapshot']}) | ||||
| expected_visit['origin_visit_url'] = origin_visit_url | expected_visit['origin_visit_url'] = origin_visit_url | ||||
| expected_visit['snapshot_url'] = snapshot_url | expected_visit['snapshot_url'] = snapshot_url | ||||
| expected_visit['origin'] = \ | |||||
| hash_to_hex(expected_visit['origin']) | |||||
| self.assertEqual(rv.data, expected_visits) | self.assertEqual(rv.data, expected_visits) | ||||
| @given(new_origin(), visit_dates(4), new_snapshots(4)) | @given(new_origin(), visit_dates(4), new_snapshots(4)) | ||||
| def test_api_lookup_origin_visit(self, new_origin, visit_dates, | def test_api_lookup_origin_visit(self, new_origin, visit_dates, | ||||
| new_snapshots): | new_snapshots): | ||||
| origin_id = self.storage.origin_add_one(new_origin) | origin_id = self.storage.origin_add_one(new_origin) | ||||
| new_origin['id'] = origin_id | new_origin['id'] = origin_id | ||||
| for i, visit_date in enumerate(visit_dates): | for i, visit_date in enumerate(visit_dates): | ||||
| origin_visit = self.storage.origin_visit_add(origin_id, visit_date) | origin_visit = self.storage.origin_visit_add(origin_id, visit_date) | ||||
| visit_id = origin_visit['visit'] | visit_id = origin_visit['visit'] | ||||
| self.storage.snapshot_add(origin_id, origin_visit['visit'], | self.storage.snapshot_add(origin_id, origin_visit['visit'], | ||||
| new_snapshots[i]) | new_snapshots[i]) | ||||
| url = reverse('api-origin-visit', | url = reverse('api-origin-visit', | ||||
| url_args={'origin_id': origin_id, | url_args={'origin_id': hash_to_hex(origin_id), | ||||
| 'visit_id': visit_id}) | 'visit_id': visit_id}) | ||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| self.assertEqual(rv.status_code, 200) | self.assertEqual(rv.status_code, 200, rv.data) | ||||
| self.assertEqual(rv['Content-Type'], 'application/json') | self.assertEqual(rv['Content-Type'], 'application/json') | ||||
| expected_visit = self.origin_visit_get_by(origin_id, visit_id) | expected_visit = self.origin_visit_get_by(origin_id, visit_id) | ||||
| origin_url = reverse('api-origin', | origin_url = reverse( | ||||
| url_args={'origin_id': origin_id}) | 'api-origin', | ||||
| url_args={'origin_id': hash_to_hex(origin_id)}) | |||||
| snapshot_url = reverse( | snapshot_url = reverse( | ||||
| 'api-snapshot', | 'api-snapshot', | ||||
| url_args={'snapshot_id': expected_visit['snapshot']}) | url_args={'snapshot_id': expected_visit['snapshot']}) | ||||
| expected_visit['origin'] = hash_to_hex(expected_visit['origin']) | |||||
| expected_visit['origin_url'] = origin_url | expected_visit['origin_url'] = origin_url | ||||
| expected_visit['snapshot_url'] = snapshot_url | expected_visit['snapshot_url'] = snapshot_url | ||||
| self.maxDiff = None | |||||
| self.assertEqual(rv.data, expected_visit) | self.assertEqual(rv.data, expected_visit) | ||||
| @given(origin()) | @given(origin()) | ||||
| def test_api_lookup_origin_visit_not_found(self, origin): | def test_api_lookup_origin_visit_not_found(self, origin): | ||||
| all_visits = list(reversed(get_origin_visits(origin))) | all_visits = list(reversed(get_origin_visits(origin))) | ||||
| max_visit_id = max([v['visit'] for v in all_visits]) | max_visit_id = max([v['visit'] for v in all_visits]) | ||||
| url = reverse('api-origin-visit', | url = reverse('api-origin-visit', | ||||
| url_args={'origin_id': origin['id'], | url_args={'origin_id': hash_to_hex(origin['id']), | ||||
| 'visit_id': max_visit_id + 1}) | 'visit_id': max_visit_id + 1}) | ||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| self.assertEqual(rv.status_code, 404) | self.assertEqual(rv.status_code, 404, rv.data) | ||||
| self.assertEqual(rv['Content-Type'], 'application/json') | self.assertEqual(rv['Content-Type'], 'application/json') | ||||
| self.assertEqual(rv.data, { | self.assertEqual(rv.data, { | ||||
| 'exception': 'NotFoundExc', | 'exception': 'NotFoundExc', | ||||
| 'reason': 'Origin with id %s or its visit with id %s not found!' % | 'reason': 'Origin with id %s or its visit with id %s not found!' % | ||||
| (origin['id'], max_visit_id+1) | (hash_to_hex(origin['id']), max_visit_id+1) | ||||
| }) | }) | ||||
| @given(origin()) | @given(origin()) | ||||
| def test_api_origin_by_id(self, origin): | def test_api_origin_by_id(self, origin): | ||||
| url = reverse('api-origin', url_args={'origin_id': origin['id']}) | url = reverse('api-origin', | ||||
| url_args={'origin_id': hash_to_hex(origin['id'])}) | |||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| expected_origin = self.origin_get(origin) | expected_origin = self.origin_get(origin) | ||||
| origin_visits_url = reverse('api-origin-visits', | origin_visits_url = reverse( | ||||
| url_args={'origin_id': origin['id']}) | 'api-origin-visits', | ||||
| url_args={'origin_id': hash_to_hex(origin['id'])}) | |||||
| expected_origin['origin_visits_url'] = origin_visits_url | expected_origin['origin_visits_url'] = origin_visits_url | ||||
| expected_origin['id'] = hash_to_hex(expected_origin['id']) | |||||
| self.assertEqual(rv.status_code, 200) | self.assertEqual(rv.status_code, 200, rv.data) | ||||
| self.assertEqual(rv['Content-Type'], 'application/json') | self.assertEqual(rv['Content-Type'], 'application/json') | ||||
| self.assertEqual(rv.data, expected_origin) | self.assertEqual(rv.data, expected_origin) | ||||
| @given(origin()) | @given(origin()) | ||||
| def test_api_origin_by_type_url(self, origin): | def test_api_origin_by_type_url(self, origin): | ||||
| url = reverse('api-origin', | url = reverse('api-origin', | ||||
| url_args={'origin_type': origin['type'], | url_args={'origin_type': origin['type'], | ||||
| 'origin_url': origin['url']}) | 'origin_url': origin['url']}) | ||||
| from swh.web.config import get_config | |||||
| get_config()['debug'] = True | |||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| get_config()['debug'] = False | |||||
| expected_origin = self.origin_get(origin) | expected_origin = self.origin_get(origin) | ||||
| origin_visits_url = reverse('api-origin-visits', | origin_visits_url = reverse( | ||||
| url_args={'origin_id': origin['id']}) | 'api-origin-visits', | ||||
| url_args={'origin_id': hash_to_hex(origin['id'])}) | |||||
| expected_origin['origin_visits_url'] = origin_visits_url | expected_origin['origin_visits_url'] = origin_visits_url | ||||
| expected_origin['id'] = hash_to_hex(expected_origin['id']) | |||||
| self.assertEqual(rv.status_code, 200) | self.assertEqual(rv.status_code, 200, rv.data) | ||||
| self.assertEqual(rv['Content-Type'], 'application/json') | self.assertEqual(rv['Content-Type'], 'application/json') | ||||
| self.assertEqual(rv.data, expected_origin) | self.assertEqual(rv.data, expected_origin) | ||||
| @given(new_origin()) | @given(new_origin()) | ||||
| def test_api_origin_not_found(self, new_origin): | def test_api_origin_not_found(self, new_origin): | ||||
| url = reverse('api-origin', | url = reverse('api-origin', | ||||
| url_args={'origin_type': new_origin['type'], | url_args={'origin_type': new_origin['type'], | ||||
| Show All 31 Lines | def test_api_origin_metadata_search(self, origin): | ||||
| url = reverse('api-origin-metadata-search', | url = reverse('api-origin-metadata-search', | ||||
| query_params={'fulltext': 'Jane Doe'}) | query_params={'fulltext': 'Jane Doe'}) | ||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| self.assertEqual(rv.status_code, 200, rv.content) | self.assertEqual(rv.status_code, 200, rv.content) | ||||
| self.assertEqual(rv['Content-Type'], 'application/json') | self.assertEqual(rv['Content-Type'], 'application/json') | ||||
| expected_data = [{ | expected_data = [{ | ||||
| 'id': origin['id'], | 'id': hash_to_hex(origin['id']), | ||||
| 'type': origin['type'], | 'type': origin['type'], | ||||
| 'url': origin['url'], | 'url': origin['url'], | ||||
| 'metadata': { | 'metadata': { | ||||
| 'metadata': {'author': 'Jane Doe'}, | 'metadata': {'author': 'Jane Doe'}, | ||||
| 'from_revision': ( | 'from_revision': ( | ||||
| '7026b7c1a2af56521e951c01ed20f255fa054238'), | '7026b7c1a2af56521e951c01ed20f255fa054238'), | ||||
| 'tool': { | 'tool': { | ||||
| 'configuration': { | 'configuration': { | ||||
| ▲ Show 20 Lines • Show All 68 Lines • ▼ Show 20 Lines | class OriginApiTestCase(WebTestCase, APITestCase): | ||||
| def test_api_origin_metadata_search_invalid(self, mock_idx_storage): | def test_api_origin_metadata_search_invalid(self, mock_idx_storage): | ||||
| url = reverse('api-origin-metadata-search') | url = reverse('api-origin-metadata-search') | ||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| self.assertEqual(rv.status_code, 400, rv.content) | self.assertEqual(rv.status_code, 400, rv.content) | ||||
| mock_idx_storage.assert_not_called() | mock_idx_storage.assert_not_called() | ||||
| @given(new_origins(20)) | @given(new_origins(10), integers(min_value=1, max_value=9)) | ||||
| def test_api_lookup_origins(self, new_origins): | def test_api_lookup_origins(self, new_origins, origin_count): | ||||
| self.maxDiff = None | |||||
| nb_origins = len(new_origins) | self.storage.origin_add(new_origins) | ||||
| expected_origins = self.storage.origin_add(new_origins) | origins = list(self.storage.origin_get_range(origin_count=10000)) | ||||
| origins.sort(key=lambda origin: origin['id']) | |||||
| nb_origins = len(origins) | |||||
| origin_from_idx = random.randint(1, nb_origins-1) - 1 | origin_from_idx = random.randint(1, nb_origins-1) - 1 | ||||
| origin_from = expected_origins[origin_from_idx]['id'] | origin_from = origins[origin_from_idx]['id'] | ||||
| max_origin_id = expected_origins[-1]['id'] | |||||
| origin_count = random.randint(1, max_origin_id - origin_from) | if origin_from_idx + origin_count >= len(origins): | ||||
| origin_count = len(origins) - origin_from_idx - 1 | |||||
| url = reverse('api-origins', | url = reverse('api-origins', | ||||
| query_params={'origin_from': origin_from, | query_params={'origin_from': hash_to_hex(origin_from), | ||||
| 'origin_count': origin_count}) | 'origin_count': origin_count}) | ||||
| rv = self.client.get(url) | rv = self.client.get(url) | ||||
| self.assertEqual(rv.status_code, 200) | self.assertEqual(rv.status_code, 200, rv.data) | ||||
| start = origin_from_idx | start = origin_from_idx | ||||
| end = origin_from_idx + origin_count | end = origin_from_idx + origin_count | ||||
| expected_origins = expected_origins[start:end] | expected_origins = origins[start:end] | ||||
| for expected_origin in expected_origins: | for expected_origin in expected_origins: | ||||
| expected_origin['id'] = hash_to_hex(expected_origin['id']) | |||||
| expected_origin['origin_visits_url'] = reverse( | expected_origin['origin_visits_url'] = reverse( | ||||
| 'api-origin-visits', | 'api-origin-visits', | ||||
| url_args={'origin_id': expected_origin['id']}) | url_args={'origin_id': expected_origin['id']}) | ||||
| self.assertEqual(rv.data, expected_origins) | self.assertEqual(rv.data, expected_origins) | ||||
| next_origin_id = expected_origins[-1]['id']+1 | next_origin_id = origins[end]['id'] | ||||
| if self.storage.origin_get({'id': next_origin_id}): | if self.storage.origin_get({'id': next_origin_id}): | ||||
| self.assertIn('Link', rv) | self.assertIn('Link', rv) | ||||
| next_url = reverse('api-origins', | next_url = reverse( | ||||
| query_params={'origin_from': next_origin_id, | 'api-origins', | ||||
| query_params={'origin_from': hash_to_hex(next_origin_id), | |||||
| 'origin_count': origin_count}) | 'origin_count': origin_count}) | ||||
| self.assertIn(next_url, rv['Link']) | self.assertIn(next_url, rv['Link']) | ||||