Changeset View
Changeset View
Standalone View
Standalone View
swh/core/db/tests/db_testing.py
| # Copyright (C) 2015-2018 The Software Heritage developers | # Copyright (C) 2015-2018 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 | ||||
| import os | import os | ||||
| import glob | import glob | ||||
| import subprocess | import subprocess | ||||
| import psycopg2 | import psycopg2 | ||||
| from typing import Dict, Iterable, Optional, Tuple, Union | |||||
| from swh.core.utils import numfile_sortkey as sortkey | from swh.core.utils import numfile_sortkey as sortkey | ||||
| DB_DUMP_TYPES = {'.sql': 'psql', '.dump': 'pg_dump'} | |||||
| DB_DUMP_TYPES = {'.sql': 'psql', '.dump': 'pg_dump'} # type: Dict[str, str] | |||||
| def swh_db_version(dbname_or_service): | def swh_db_version(dbname_or_service): | ||||
| """Retrieve the swh version if any. In case of the db not initialized, | """Retrieve the swh version if any. In case of the db not initialized, | ||||
| this returns None. Otherwise, this returns the db's version. | this returns None. Otherwise, this returns the db's version. | ||||
| Args: | Args: | ||||
| dbname_or_service (str): The db's name or service | dbname_or_service (str): The db's name or service | ||||
| ▲ Show 20 Lines • Show All 179 Lines • ▼ Show 20 Lines | defined in a test case: | ||||
| @classmethod | @classmethod | ||||
| def tearDownClass(cls): | def tearDownClass(cls): | ||||
| # your class teardown code here | # your class teardown code here | ||||
| super().tearDownClass() | super().tearDownClass() | ||||
| """ | """ | ||||
| _DB_DUMP_LIST = {} | _DB_DUMP_LIST = {} # type: Dict[str, Iterable[Tuple[str, str]]] | ||||
| _DB_LIST = {} | _DB_LIST = {} # type: Dict[str, DbTestContext] | ||||
| DB_TEST_FIXTURE_IMPORTED = True | DB_TEST_FIXTURE_IMPORTED = True | ||||
| @classmethod | @classmethod | ||||
| def add_db(cls, name='softwareheritage-test', dumps=None): | def add_db(cls, name='softwareheritage-test', dumps=None): | ||||
| cls._DB_DUMP_LIST[name] = dumps | cls._DB_DUMP_LIST[name] = dumps | ||||
| @classmethod | @classmethod | ||||
| def setUpClass(cls): | def setUpClass(cls): | ||||
| ▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines | class SingleDbTestFixture(DbTestFixture): | ||||
| self: | self: | ||||
| dbname: name of the test database | dbname: name of the test database | ||||
| conn: psycopg2 connection object | conn: psycopg2 connection object | ||||
| cursor: open psycopg2 cursor to the DB | cursor: open psycopg2 cursor to the DB | ||||
| """ | """ | ||||
| TEST_DB_NAME = 'softwareheritage-test' | TEST_DB_NAME = 'softwareheritage-test' | ||||
| TEST_DB_DUMP = None | TEST_DB_DUMP = None # type: Optional[Union[str, Iterable[str]]] | ||||
| @classmethod | @classmethod | ||||
| def setUpClass(cls): | def setUpClass(cls): | ||||
| cls.dbname = cls.TEST_DB_NAME # XXX to kill? | cls.dbname = cls.TEST_DB_NAME # XXX to kill? | ||||
| dump_files = cls.TEST_DB_DUMP | dump_files = cls.TEST_DB_DUMP | ||||
| if dump_files is None: | if dump_files is None: | ||||
| dump_files = [] | dump_files = [] | ||||
| Show All 20 Lines | |||||