Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| # Copyright (C) 2015-2019 The Software Heritage developers | # Copyright (C) 2015-2019 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 | ||||
| import copy | import copy | ||||
| from contextlib import contextmanager | from contextlib import contextmanager | ||||
| import datetime | import datetime | ||||
| import itertools | import itertools | ||||
| import queue | import queue | ||||
| import threading | import threading | ||||
| from collections import defaultdict | from collections import defaultdict | ||||
| from unittest.mock import Mock, patch | from unittest.mock import Mock | ||||
| import psycopg2 | import psycopg2 | ||||
| import pytest | import pytest | ||||
| from hypothesis import given, strategies, settings, HealthCheck | from hypothesis import given, strategies, settings, HealthCheck | ||||
| from typing import ClassVar, Optional | from typing import ClassVar, Optional | ||||
| ▲ Show 20 Lines • Show All 3,247 Lines • ▼ Show 20 Lines | def test_add_arbitrary(self, swh_storage, objects): | ||||
| pass | pass | ||||
| @pytest.mark.db | @pytest.mark.db | ||||
| class TestLocalStorage: | class TestLocalStorage: | ||||
| """Test the local storage""" | """Test the local storage""" | ||||
| _test_origin_ids = True | _test_origin_ids = True | ||||
| # Can only be tested with local storage as you can't mock | |||||
| # datetimes for the remote server | |||||
| @pytest.mark.parametrize('use_url', [True, False]) | |||||
| def test_fetch_history(self, swh_storage, use_url): | |||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| origin_id = swh_storage.origin_add_one(data.origin) | |||||
| origin_id_or_url = data.origin['url'] if use_url else origin_id | |||||
| with patch('datetime.datetime'): | |||||
| datetime.datetime.now.return_value = data.fetch_history_date | |||||
| fetch_history_id = swh_storage.fetch_history_start( | |||||
| origin_id_or_url) | |||||
| datetime.datetime.now.assert_called_with(tz=datetime.timezone.utc) | |||||
| with patch('datetime.datetime'): | |||||
| datetime.datetime.now.return_value = data.fetch_history_end | |||||
| swh_storage.fetch_history_end(fetch_history_id, | |||||
| data.fetch_history_data) | |||||
| fetch_history = swh_storage.fetch_history_get(fetch_history_id) | |||||
| expected_fetch_history = data.fetch_history_data.copy() | |||||
| expected_fetch_history['id'] = fetch_history_id | |||||
| expected_fetch_history['origin'] = origin_id | |||||
| expected_fetch_history['date'] = data.fetch_history_date | |||||
| expected_fetch_history['duration'] = data.fetch_history_duration | |||||
| assert expected_fetch_history == fetch_history | |||||
| # This test is only relevant on the local storage, with an actual | # This test is only relevant on the local storage, with an actual | ||||
| # objstorage raising an exception | # objstorage raising an exception | ||||
| def test_content_add_objstorage_exception(self, swh_storage): | def test_content_add_objstorage_exception(self, swh_storage): | ||||
| swh_storage.objstorage.add = Mock( | swh_storage.objstorage.add = Mock( | ||||
| side_effect=Exception('mocked broken objstorage') | side_effect=Exception('mocked broken objstorage') | ||||
| ) | ) | ||||
| with pytest.raises(Exception) as e: | with pytest.raises(Exception) as e: | ||||
| ▲ Show 20 Lines • Show All 185 Lines • Show Last 20 Lines | |||||