Skip to content
Snippets Groups Projects

vault.tests: Make postgresql fixture faster

Compare and
2 files
+ 21
30
Compare changes
  • Side-by-side
  • Inline
Files
2
+ 20
29
import glob
# Copyright (C) 2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
import os
import subprocess
from typing import Any, Dict
import pkg_resources.extern.packaging.version
import pytest
from pytest_postgresql import factories
from swh.core.utils import numfile_sortkey as sortkey
from swh.core.db.pytest_plugin import postgresql_fact
from swh.storage.tests import SQL_DIR as STORAGE_SQL_DIR
import swh.vault
from swh.vault import get_vault
from swh.vault.tests import SQL_DIR
os.environ["LC_ALL"] = "C.UTF-8"
@@ -35,17 +37,25 @@ def db_url(name, postgresql_proc):
)
postgresql2 = factories.postgresql("postgresql_proc", "tests2")
VAULT_SQL_DIR = os.path.join(os.path.dirname(swh.vault.__file__), "sql")
postgres_vault = postgresql_fact(
"postgresql_proc", db_name="vault", dump_files=f"{VAULT_SQL_DIR}/*.sql"
)
postgres_storage = postgresql_fact(
"postgresql_proc", db_name="storage", dump_files=f"{STORAGE_SQL_DIR}/*.sql"
)
@pytest.fixture
def swh_vault_config(postgresql, postgresql2, tmp_path) -> Dict[str, Any]:
def swh_vault_config(postgres_vault, postgres_storage, tmp_path) -> Dict[str, Any]:
tmp_path = str(tmp_path)
return {
"db": postgresql.dsn,
"db": postgres_vault.dsn,
"storage": {
"cls": "local",
"db": postgresql2.dsn,
"db": postgres_storage.dsn,
"objstorage": {
"cls": "pathslicing",
"args": {"root": tmp_path, "slicing": "0:1/1:5",},
@@ -60,26 +70,7 @@ def swh_vault_config(postgresql, postgresql2, tmp_path) -> Dict[str, Any]:
@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)):
dump_files = os.path.join(sql_dir, "*.sql")
all_dump_files = sorted(glob.glob(dump_files), key=sortkey)
for fname in all_dump_files:
subprocess.check_call(
[
"psql",
"--quiet",
"--no-psqlrc",
"-v",
"ON_ERROR_STOP=1",
"-d",
pg.dsn,
"-f",
fname,
]
)
def swh_vault(request, swh_vault_config):
return get_vault("local", **swh_vault_config)
Loading