Differential D816 Diff 2602 swh/core/tests/db_testing.py

Changeset View

Standalone View

swh/core/tests/db_testing.py

# Copyright (C) 2015 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 psycopg2		import psycopg2
import subprocess		import subprocess

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'}


		def swh_db_version(dbname_or_service):
		"""Retrieve the swh version if any. In case of the db not initialized,
		this returns None. Otherwise, this returns the db's version.

		Args:
		dbname_or_service (str): The db's name or service

		Returns:
		Optional[Int]: Either the db's version or None

		"""
		query = 'select version from dbversion order by dbversion desc limit 1'
		cmd = [
		'psql', '--tuples-only', '--no-psqlrc', '--quiet',
		'-v', 'ON_ERROR_STOP=1', "--command=%s" % query,
		dbname_or_service,
		]

		print('cmd: %s' % cmd)

		try:
		r = subprocess.run(cmd, check=True, stdout=subprocess.PIPE,
		universal_newlines=True)
		result = int(r.stdout.strip())
		vlorentzUnsubmitted Not Done Inline Actions Can't we use psycopg2 for that? vlorentz: Can't we use psycopg2 for that?
		ardumontAuthorUnsubmitted Done Inline Actions I tried but then that would mean i'd need to change the input. As dbname is not enough. At the moment, we can both use dbname and a string like 'service=swh' (providing you have the right .pgass and .pg_service.conf installed). ardumont: I tried but then that would mean i'd need to change the input. As dbname is not enough. At the…
		ardumontAuthorUnsubmitted Done Inline Actions What i meant was, in the current context it's used (swh.core.cli), this is fine. If we want to actually implement using psycopg2, we will have to draw a not so simple refactoring (including the swh.core.cli which is not tested yet). ardumont: What i meant was, in the current context it's used (swh.core.cli), this is fine. If we want to…
		except Exception: # db not initialized
		result = None
		return result


def pg_restore(dbname, dumpfile, dumptype='pg_dump'):		def pg_restore(dbname, dumpfile, dumptype='pg_dump'):
"""		"""
Args:		Args:
dbname: name of the DB to restore into		dbname: name of the DB to restore into
dumpfile: path fo the dump file		dumpfile: path fo the dump file
dumptype: one of 'pg_dump' (for binary dumps), 'psql' (for SQL dumps)		dumptype: one of 'pg_dump' (for binary dumps), 'psql' (for SQL dumps)
"""		"""
assert dumptype in ['pg_dump', 'psql']		assert dumptype in ['pg_dump', 'psql']
Show All 12 Lines	def pg_dump(dbname, dumpfile):
subprocess.check_call(['pg_dump', '--no-owner', '--no-privileges', '-Fc',		subprocess.check_call(['pg_dump', '--no-owner', '--no-privileges', '-Fc',
'-f', dumpfile, dbname])		'-f', dumpfile, dbname])


def pg_dropdb(dbname):		def pg_dropdb(dbname):
subprocess.check_call(['dropdb', dbname])		subprocess.check_call(['dropdb', dbname])


def pg_createdb(dbname):		def pg_createdb(dbname, check=True):
subprocess.check_call(['createdb', dbname])		"""Create a db. If check is True and the db already exists, this will
		raise an exception (original behavior). If check is False and
		the db already exists, this will fail silently. If the db does
		not exist, the db will be created.

		"""
		_run = subprocess.check_call if check else subprocess.call
		vlorentzUnsubmitted Not Done Inline Actions Starting with Python 3.5, you can do this instead: subprocess.run(['createdb', dbname], check=check) vlorentz: Starting with Python 3.5, you can do this instead: ``` subprocess.run(['createdb', dbname]…
		ardumontAuthorUnsubmitted Done Inline Actions So true! ardumont: So true!
		_run(['createdb', dbname])


def db_create(dbname, dumps=None):		def db_create(dbname, dumps=None):
"""create the test DB and load the test data dumps into it		"""create the test DB and load the test data dumps into it

dumps is an iterable of couples (dump_file, dump_type).		dumps is an iterable of couples (dump_file, dump_type).

context: setUpClass		context: setUpClass
▲ Show 20 Lines • Show All 228 Lines • Show Last 20 Lines