Changeset View
Changeset View
Standalone View
Standalone View
swh/vault/tests/test_backend.py
# Copyright (C) 2017 The Software Heritage developers | # Copyright (C) 2017-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 | ||||
import contextlib | import contextlib | ||||
import datetime | import datetime | ||||
from unittest.mock import MagicMock, patch | from unittest.mock import MagicMock, patch | ||||
import attr | |||||
import psycopg2 | import psycopg2 | ||||
import pytest | import pytest | ||||
from swh.model import hashutil | from swh.model import hashutil | ||||
from swh.vault.exc import NotFoundExc | |||||
from swh.vault.tests.vault_testing import hash_content | from swh.vault.tests.vault_testing import hash_content | ||||
@contextlib.contextmanager | @contextlib.contextmanager | ||||
def mock_cooking(vault_backend): | def mock_cooking(vault_backend): | ||||
with patch.object(vault_backend, "_send_task") as mt: | with patch.object(vault_backend, "_send_task") as mt: | ||||
mt.return_value = 42 | mt.return_value = 42 | ||||
with patch("swh.vault.backend.get_cooker_cls") as mg: | with patch("swh.vault.backend.get_cooker_cls") as mg: | ||||
Show All 37 Lines | |||||
TEST_HEX_ID = "4a4b9771542143cf070386f86b4b92d42966bdbc" | TEST_HEX_ID = "4a4b9771542143cf070386f86b4b92d42966bdbc" | ||||
TEST_OBJ_ID = hashutil.hash_to_bytes(TEST_HEX_ID) | TEST_OBJ_ID = hashutil.hash_to_bytes(TEST_HEX_ID) | ||||
TEST_PROGRESS = ( | TEST_PROGRESS = ( | ||||
"Mr. White, You're telling me you're cooking again?" " \N{ASTONISHED FACE} " | "Mr. White, You're telling me you're cooking again?" " \N{ASTONISHED FACE} " | ||||
) | ) | ||||
TEST_EMAIL = "ouiche@lorraine.fr" | TEST_EMAIL = "ouiche@lorraine.fr" | ||||
@pytest.fixture | |||||
def swh_vault(swh_vault, sample_data): | |||||
# make the vault's storage consistent with test data | |||||
revision = attr.evolve(sample_data.revision, id=TEST_OBJ_ID) | |||||
swh_vault.storage.revision_add([revision]) | |||||
return swh_vault | |||||
def test_create_task_simple(swh_vault): | def test_create_task_simple(swh_vault): | ||||
with mock_cooking(swh_vault) as m: | with mock_cooking(swh_vault) as m: | ||||
swh_vault.create_task(TEST_TYPE, TEST_OBJ_ID) | swh_vault.create_task(TEST_TYPE, TEST_OBJ_ID) | ||||
m["get_cooker_cls"].assert_called_once_with(TEST_TYPE) | m["get_cooker_cls"].assert_called_once_with(TEST_TYPE) | ||||
args = m["cooker_cls"].call_args[0] | args = m["cooker_cls"].call_args[0] | ||||
assert args[0] == TEST_TYPE | assert args[0] == TEST_TYPE | ||||
▲ Show 20 Lines • Show All 79 Lines • ▼ Show 20 Lines | def test_create_update_access_ts(swh_vault): | ||||
info = swh_vault.task_info(TEST_TYPE, TEST_OBJ_ID) | info = swh_vault.task_info(TEST_TYPE, TEST_OBJ_ID) | ||||
access_ts_3 = info["ts_last_access"] | access_ts_3 = info["ts_last_access"] | ||||
assertTimestampAlmostNow(access_ts_3) | assertTimestampAlmostNow(access_ts_3) | ||||
assert access_ts_1 < access_ts_2 | assert access_ts_1 < access_ts_2 | ||||
assert access_ts_2 < access_ts_3 | assert access_ts_2 < access_ts_3 | ||||
def test_cook_request_idempotent(swh_vault): | def test_cook_request_idempotent(swh_vault, sample_data): | ||||
with mock_cooking(swh_vault): | with mock_cooking(swh_vault): | ||||
info1 = swh_vault.cook_request(TEST_TYPE, TEST_OBJ_ID) | info1 = swh_vault.cook_request(TEST_TYPE, TEST_OBJ_ID) | ||||
info2 = swh_vault.cook_request(TEST_TYPE, TEST_OBJ_ID) | info2 = swh_vault.cook_request(TEST_TYPE, TEST_OBJ_ID) | ||||
info3 = swh_vault.cook_request(TEST_TYPE, TEST_OBJ_ID) | info3 = swh_vault.cook_request(TEST_TYPE, TEST_OBJ_ID) | ||||
assert info1 == info2 | assert info1 == info2 | ||||
assert info1 == info3 | assert info1 == info3 | ||||
▲ Show 20 Lines • Show All 62 Lines • ▼ Show 20 Lines | def test_available(swh_vault): | ||||
swh_vault.cache.add(TEST_TYPE, TEST_OBJ_ID, b"content") | swh_vault.cache.add(TEST_TYPE, TEST_OBJ_ID, b"content") | ||||
assert not swh_vault.is_available(TEST_TYPE, TEST_OBJ_ID) | assert not swh_vault.is_available(TEST_TYPE, TEST_OBJ_ID) | ||||
swh_vault.set_status(TEST_TYPE, TEST_OBJ_ID, "done") | swh_vault.set_status(TEST_TYPE, TEST_OBJ_ID, "done") | ||||
assert swh_vault.is_available(TEST_TYPE, TEST_OBJ_ID) | assert swh_vault.is_available(TEST_TYPE, TEST_OBJ_ID) | ||||
def test_fetch(swh_vault): | def test_fetch(swh_vault): | ||||
assert swh_vault.fetch(TEST_TYPE, TEST_OBJ_ID) is None | assert swh_vault.fetch(TEST_TYPE, TEST_OBJ_ID, raise_notfound=False) is None | ||||
with pytest.raises( | |||||
NotFoundExc, match=f"{TEST_TYPE} {TEST_HEX_ID} is not available." | |||||
): | |||||
swh_vault.fetch(TEST_TYPE, TEST_OBJ_ID) | |||||
obj_id, content = fake_cook(swh_vault, TEST_TYPE, b"content") | obj_id, content = fake_cook(swh_vault, TEST_TYPE, b"content") | ||||
info = swh_vault.task_info(TEST_TYPE, obj_id) | info = swh_vault.task_info(TEST_TYPE, obj_id) | ||||
access_ts_before = info["ts_last_access"] | access_ts_before = info["ts_last_access"] | ||||
assert swh_vault.fetch(TEST_TYPE, obj_id) == b"content" | assert swh_vault.fetch(TEST_TYPE, obj_id) == b"content" | ||||
info = swh_vault.task_info(TEST_TYPE, obj_id) | info = swh_vault.task_info(TEST_TYPE, obj_id) | ||||
▲ Show 20 Lines • Show All 87 Lines • Show Last 20 Lines |