Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show All 16 Lines | |||||
from hypothesis import given, strategies, settings, HealthCheck | from hypothesis import given, strategies, settings, HealthCheck | ||||
from typing import ClassVar, Optional | from typing import ClassVar, Optional | ||||
from swh.model import from_disk, identifiers | from swh.model import from_disk, identifiers | ||||
from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
from swh.model.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
from swh.storage.tests.storage_testing import StorageTestFixture | |||||
from swh.storage import HashCollision | from swh.storage import HashCollision | ||||
from .storage_data import data | from .storage_data import data | ||||
@pytest.mark.db | |||||
class StorageTestDbFixture(StorageTestFixture): | |||||
def setUp(self): | |||||
super().setUp() | |||||
self.maxDiff = None | |||||
def tearDown(self): | |||||
self.reset_storage() | |||||
if hasattr(self.storage, '_pool') and self.storage._pool: | |||||
self.storage._pool.closeall() | |||||
super().tearDown() | |||||
def get_db(self): | |||||
return self.storage.db() | |||||
@contextmanager | |||||
def db_transaction(self): | |||||
with self.get_db() as db: | |||||
with db.transaction() as cur: | |||||
yield db, cur | |||||
@contextmanager | @contextmanager | ||||
def db_transaction(storage): | def db_transaction(storage): | ||||
with storage.db() as db: | with storage.db() as db: | ||||
with db.transaction() as cur: | with db.transaction() as cur: | ||||
yield db, cur | yield db, cur | ||||
def normalize_entity(entity): | def normalize_entity(entity): | ||||
▲ Show 20 Lines • Show All 3,462 Lines • Show Last 20 Lines |