Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_postgresql.py
| # Copyright (C) 2015-2020 The Software Heritage developers | # Copyright (C) 2015-2022 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 contextlib import contextmanager | from contextlib import contextmanager | ||||
| import queue | import queue | ||||
| import threading | import threading | ||||
| from unittest.mock import Mock | from unittest.mock import Mock | ||||
| import attr | import attr | ||||
| import pytest | import pytest | ||||
| from swh.model.model import Person | from swh.model.model import Person | ||||
| from swh.storage.postgresql.db import Db | |||||
| from swh.storage.tests.storage_tests import TestStorage as _TestStorage | from swh.storage.tests.storage_tests import TestStorage as _TestStorage | ||||
| from swh.storage.tests.storage_tests import TestStorageGeneratedData # noqa | from swh.storage.tests.storage_tests import TestStorageGeneratedData # noqa | ||||
| from swh.storage.utils import now | from swh.storage.utils import now | ||||
| @contextmanager | @contextmanager | ||||
| def db_transaction(storage): | def db_transaction(storage): | ||||
| with storage.db() as db: | with storage.db() as db: | ||||
| ▲ Show 20 Lines • Show All 362 Lines • ▼ Show 20 Lines | class TestPgStorage: | ||||
| def test_clear_buffers(self, swh_storage): | def test_clear_buffers(self, swh_storage): | ||||
| """Calling clear buffers on real storage does nothing""" | """Calling clear buffers on real storage does nothing""" | ||||
| assert swh_storage.clear_buffers() is None | assert swh_storage.clear_buffers() is None | ||||
| def test_flush(self, swh_storage): | def test_flush(self, swh_storage): | ||||
| """Calling clear buffers on real storage does nothing""" | """Calling clear buffers on real storage does nothing""" | ||||
| assert swh_storage.flush() == {} | assert swh_storage.flush() == {} | ||||
| def test_dbversion(self, swh_storage): | |||||
| with swh_storage.db() as db: | |||||
| assert db.check_dbversion() | |||||
| def test_dbversion_mismatch(self, swh_storage, monkeypatch): | |||||
| monkeypatch.setattr(Db, "current_version", -1) | |||||
| with swh_storage.db() as db: | |||||
| assert db.check_dbversion() is False | |||||
| def test_check_config(self, swh_storage): | def test_check_config(self, swh_storage): | ||||
| assert swh_storage.check_config(check_write=True) | assert swh_storage.check_config(check_write=True) | ||||
| assert swh_storage.check_config(check_write=False) | assert swh_storage.check_config(check_write=False) | ||||
| def test_check_config_dbversion(self, swh_storage, monkeypatch): | def test_check_config_dbversion(self, swh_storage, monkeypatch): | ||||
| monkeypatch.setattr(Db, "current_version", -1) | swh_storage.current_version = -1 | ||||
| assert swh_storage.check_config(check_write=True) is False | assert swh_storage.check_config(check_write=True) is False | ||||
| assert swh_storage.check_config(check_write=False) is False | assert swh_storage.check_config(check_write=False) is False | ||||