Changeset View
Changeset View
Standalone View
Standalone View
swh/vault/tests/conftest.py
import glob | import glob | ||||
import os | import os | ||||
import subprocess | import subprocess | ||||
from typing import Any, Dict | |||||
import pkg_resources.extern.packaging.version | import pkg_resources.extern.packaging.version | ||||
import pytest | import pytest | ||||
from pytest_postgresql import factories | from pytest_postgresql import factories | ||||
from swh.core.utils import numfile_sortkey as sortkey | from swh.core.utils import numfile_sortkey as sortkey | ||||
from swh.storage.tests import SQL_DIR as STORAGE_SQL_DIR | from swh.storage.tests import SQL_DIR as STORAGE_SQL_DIR | ||||
from swh.vault import get_vault | from swh.vault import get_vault | ||||
Show All 21 Lines | return "postgresql://{user}@{host}:{port}/{dbname}".format( | ||||
dbname=name, | dbname=name, | ||||
) | ) | ||||
postgresql2 = factories.postgresql("postgresql_proc", "tests2") | postgresql2 = factories.postgresql("postgresql_proc", "tests2") | ||||
@pytest.fixture | @pytest.fixture | ||||
def swh_vault(request, postgresql_proc, postgresql, postgresql2, tmp_path): | def swh_vault_config(postgresql, postgresql2, tmp_path) -> Dict[str, Any]: | ||||
tmp_path = str(tmp_path) | |||||
return { | |||||
"db": postgresql.dsn, | |||||
"storage": { | |||||
"cls": "local", | |||||
"db": postgresql2.dsn, | |||||
"objstorage": { | |||||
"cls": "pathslicing", | |||||
"args": {"root": tmp_path, "slicing": "0:1/1:5",}, | |||||
}, | |||||
}, | |||||
"cache": { | |||||
"cls": "pathslicing", | |||||
"args": {"root": tmp_path, "slicing": "0:1/1:5", "allow_delete": True,}, | |||||
}, | |||||
"scheduler": {"cls": "remote", "url": "http://swh-scheduler:5008",}, | |||||
} | |||||
@pytest.fixture | |||||
def swh_vault(request, swh_vault_config, postgresql, postgresql2, tmp_path): | |||||
for sql_dir, pg in ((SQL_DIR, postgresql), (STORAGE_SQL_DIR, postgresql2)): | for sql_dir, pg in ((SQL_DIR, postgresql), (STORAGE_SQL_DIR, postgresql2)): | ||||
dump_files = os.path.join(sql_dir, "*.sql") | dump_files = os.path.join(sql_dir, "*.sql") | ||||
all_dump_files = sorted(glob.glob(dump_files), key=sortkey) | all_dump_files = sorted(glob.glob(dump_files), key=sortkey) | ||||
for fname in all_dump_files: | for fname in all_dump_files: | ||||
subprocess.check_call( | subprocess.check_call( | ||||
[ | [ | ||||
"psql", | "psql", | ||||
"--quiet", | "--quiet", | ||||
"--no-psqlrc", | "--no-psqlrc", | ||||
"-v", | "-v", | ||||
"ON_ERROR_STOP=1", | "ON_ERROR_STOP=1", | ||||
"-d", | "-d", | ||||
pg.dsn, | pg.dsn, | ||||
"-f", | "-f", | ||||
fname, | fname, | ||||
] | ] | ||||
) | ) | ||||
vault_config = { | return get_vault("local", **swh_vault_config) | ||||
"db": db_url("tests", postgresql_proc), | |||||
"storage": { | |||||
"cls": "local", | |||||
"db": db_url("tests2", postgresql_proc), | |||||
"objstorage": { | |||||
"cls": "pathslicing", | |||||
"args": {"root": str(tmp_path), "slicing": "0:1/1:5",}, | |||||
}, | |||||
}, | |||||
"cache": { | |||||
"cls": "pathslicing", | |||||
"args": { | |||||
"root": str(tmp_path), | |||||
"slicing": "0:1/1:5", | |||||
"allow_delete": True, | |||||
}, | |||||
}, | |||||
"scheduler": {"cls": "remote", "url": "http://swh-scheduler:5008",}, | |||||
} | |||||
return get_vault("local", vault_config) | |||||
@pytest.fixture | @pytest.fixture | ||||
def swh_storage(swh_vault): | def swh_storage(swh_vault): | ||||
return swh_vault.storage | return swh_vault.storage |