Differential D4294 Diff 15176 swh/vault/tests/conftest.py

Changeset 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": {"args": {"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