diff6
/
python-scripttease
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Updated pgsql overlay for consistency.

Browse Source
development
Shawn Davis 4 years ago
parent a0494ec4d3
commit 4f2bb51edc
5 changed files with 128 additions and 132 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      VERSION.txt
  2. 163
      scripttease/library/overlays/pgsql.py
  3. 8
      scripttease/version.py
  4. 15
      tests/examples/kitchen_sink.ini
  5. 60
      tests/test_library_overlays_pgsql.py

2
VERSION.txt
Unescape View File

@ -1 +1 @@
6.4.4-d
6.5.0-d

163
scripttease/library/overlays/pgsql.py
Unescape View File

@ -6,13 +6,12 @@ from ..commands import Command
__all__ = (
"PGSQL_MAPPINGS",
"pg_create_database",
"pg_create_user",
"pg_database_exists",
"pg_drop_database",
"pg_drop_user",
"pg_dump_database",
"psql",
"pgsql_create",
"pgsql_drop",
"pgsql_dump",
"pgsql_exec",
"pgsql_exists",
"pgsql_user",
)
# Functions
@ -52,11 +51,11 @@ def _get_pgsql_command(name, host="localhost", password=None, port=5432, user="p
return a
def pg_create_database(name, admin_pass=None, admin_user="postgres", host="localhost", owner=None, port=5432,
template=None, **kwargs):
def pgsql_create(database, admin_pass=None, admin_user="postgres", host="localhost", owner=None, port=5432, template=None,
**kwargs):
"""Create a PostgreSQL database.
- name (str): The database name.
- database (str): The database name.
- admin_pass (str): The password for the user with sufficient access privileges to execute the command.
- admin_user (str): The name of the user with sufficient access privileges to execute the command.
- host (str): The database host name or IP address.
@ -77,15 +76,15 @@ def pg_create_database(name, admin_pass=None, admin_user="postgres", host="local
if template is not None:
base.append("--template=%s" % template)
base.append(name)
base.append(database)
return Command(" ".join(base), **kwargs)
def pg_create_user(name, admin_pass=None, admin_user="postgres", host="localhost", password=None, port=5432, **kwargs):
"""Create a PostgreSQL user.
def pgsql_drop(database, admin_pass=None, admin_user="postgres", host="localhost", port=5432, **kwargs):
"""Remove a PostgreSQL database.
- name (str): The user name.
- database (str): The database name.
- admin_pass (str): The password for the user with sufficient access privileges to execute the command.
- admin_user (str): The name of the user with sufficient access privileges to execute the command.
- host (str): The database host name or IP address.
@ -96,44 +95,42 @@ def pg_create_user(name, admin_pass=None, admin_user="postgres", host="localhost
kwargs['sudo'] = False
# Assemble the command.
base = _get_pgsql_command("createuser", host=host, password=admin_pass, port=port)
base.append("-DRS")
base.append(name)
if password is not None:
base.append("&& psql -h %s -U %s" % (host, admin_user))
base.append("-c \"ALTER USER %s WITH ENCRYPTED PASSWORD '%s';\"" % (name, password))
base = _get_pgsql_command("dropdb", host=host, password=admin_pass, port=port, user=admin_user)
base.append(database)
return Command(" ".join(base), **kwargs)
def pg_database_exists(name, admin_pass=None, admin_user="postgres", host="localhost", port=5432, **kwargs):
"""Determine if a Postgres database exists.
def pgsql_dump(database, admin_pass=None, admin_user="postgres", file_name=None, host="localhost", port=5432, **kwargs):
"""Export a Postgres database.
- name (str): The database name.
- database (str): The database name.
- admin_pass (str): The password for the user with sufficient access privileges to execute the command.
- admin_user (str): The name of the user with sufficient access privileges to execute the command.
- file_name (str): The name/path of the export file. Defaults the database name plus ``.sql``.
- host (str): The database host name or IP address.
- owner (str): The owner (user/role name) of the new database.
- port (int): The port number of the Postgres service running on the host.
"""
# Postgres commands always run without sudo because the -U may be provided. However, sudo may be required for
# file writing.
# kwargs['sudo'] = False
_file_name = file_name or "%s.sql" % database
kwargs.setdefault("register", "%s_db_exists" % name)
# Postgres commands always run without sudo because the -U may be provided.
# kwargs['sudo'] = False
base = _get_pgsql_command("psql", host=host, password=admin_pass, port=port, user=admin_user)
base.append(r"-lqt | cut -d \| -f 1 | grep -qw %s" % name)
# Assemble the command.
base = _get_pgsql_command("pg_dump", host=host, password=admin_pass, port=port, user=admin_user)
base.append("--column-inserts")
base.append("--file=%s" % _file_name)
base.append(database)
return Command(" ".join(base), **kwargs)
def pg_drop_database(name, admin_pass=None, admin_user="postgres", host="localhost", port=5432, **kwargs):
"""Remove a PostgreSQL database.
def pgsql_exec(sql, database="template1", host="localhost", password=None, port=5432, user="postgres", **kwargs):
"""Execute a psql command.
- name (str): The database name.
- sql (str): The SQL to be executed.
- database (str): The database name.
- admin_pass (str): The password for the user with sufficient access privileges to execute the command.
- admin_user (str): The name of the user with sufficient access privileges to execute the command.
- host (str): The database host name or IP address.
@ -144,94 +141,86 @@ def pg_drop_database(name, admin_pass=None, admin_user="postgres", host="localho
kwargs['sudo'] = False
# Assemble the command.
base = _get_pgsql_command("dropdb", host=host, password=admin_pass, port=port, user=admin_user)
base.append(name)
base = _get_pgsql_command("psql", host=host, password=password, port=port, user=user)
base.append("--dbname=%s" % database)
base.append('-c "%s"' % sql)
return Command(" ".join(base), **kwargs)
def pg_drop_user(name, admin_pass=None, admin_user="postgres", host="localhost", port=5432, **kwargs):
"""Remove a Postgres user.
def pgsql_exists(database, admin_pass=None, admin_user="postgres", host="localhost", port=5432, **kwargs):
"""Determine if a Postgres database exists.
- name (str): The user name.
- database (str): The database name.
- admin_pass (str): The password for the user with sufficient access privileges to execute the command.
- admin_user (str): The name of the user with sufficient access privileges to execute the command.
- host (str): The database host name or IP address.
- owner (str): The owner (user/role name) of the new database.
- port (int): The port number of the Postgres service running on the host.
"""
# Postgres commands always run without sudo because the -U may be provided.
kwargs['sudo'] = False
kwargs.setdefault("register", "pgsql_db_exists")
# Assemble the command.
base = _get_pgsql_command("dropuser", host=host, password=admin_pass, port=port, user=admin_user)
base.append(name)
base = _get_pgsql_command("psql", host=host, password=admin_pass, port=port, user=admin_user)
base.append(r"-lqt | cut -d \| -f 1 | grep -qw %s" % database)
return Command(" ".join(base), **kwargs)
def pg_dump_database(name, admin_pass=None, admin_user="postgres", file_name=None, host="localhost", port=5432,
**kwargs):
"""Export a Postgres database.
def pgsql_user(name, admin_pass=None, admin_user="postgres", host="localhost", op="create", password=None, port=5432, **kwargs):
"""Work with a PostgreSQL user.
- name (str): The database name.
- admin_pass (str): The password for the user with sufficient access privileges to execute the command.
- admin_user (str): The name of the user with sufficient access privileges to execute the command.
- file_name (str): The name/path of the export file. Defaults the database name plus ``.sql``.
- host (str): The database host name or IP address.
- port (int): The port number of the Postgres service running on the host.
- name (str): The user name.
- host (str): The host name.
- op (str): The operation to perform: ``create``, ``drop``, ``exists``.
- passwd (str): The password for a new user.
- password (str): The password for the user with sufficient access privileges to execute the command.
- port (int): The TCP port number.
- user (str): The name of the user with sufficient access privileges to execute the command.
"""
_file_name = file_name or "%s.sql" % name
# Postgres commands always run without sudo because the -U may be provided.
kwargs['sudo'] = False
if op == "create":
kwargs.setdefault("comment", "create %s postgres user" % name)
# Assemble the command.
base = _get_pgsql_command("pg_dump", host=host, password=admin_pass, port=port, user=admin_user)
base.append("--column-inserts")
base.append("--file=%s" % _file_name)
base = _get_pgsql_command("createuser", host=host, password=admin_pass, port=port)
base.append("-DRS")
base.append(name)
return Command(" ".join(base), **kwargs)
if password is not None:
base.append("&& psql -h %s -U %s" % (host, admin_user))
base.append("-c \"ALTER USER %s WITH ENCRYPTED PASSWORD '%s';\"" % (name, password))
def psql(sql, database="template1", host="localhost", password=None, port=5432, user="postgres", **kwargs):
"""Execute a psql command.
return Command(" ".join(base), **kwargs)
elif op == "drop":
kwargs.setdefault("comment", "drop %s postgres user" % name)
base = _get_pgsql_command("dropuser", host=host, password=admin_pass, port=port, user=admin_user)
base.append(name)
- sql (str): The SQL to be executed.
- database (str): The database name.
- admin_pass (str): The password for the user with sufficient access privileges to execute the command.
- admin_user (str): The name of the user with sufficient access privileges to execute the command.
- host (str): The database host name or IP address.
- port (int): The port number of the Postgres service running on the host.
return Command(" ".join(base), **kwargs)
elif op == "exists":
kwargs.setdefault("comment", "determine if %s postgres user exits" % name)
kwargs.setdefault("register", "pgsql_use_exists")
"""
# Postgres commands always run without sudo because the -U may be provided.
kwargs['sudo'] = False
base = _get_pgsql_command("psql", host=host, password=admin_pass, port=port, user=admin_user)
# Assemble the command.
base = _get_pgsql_command("psql", host=host, password=password, port=port, user=user)
base.append("--dbname=%s" % database)
sql = "SELECT 1 FROM pgsql_roles WHERE rolname='%s'" % name
base.append('-c "%s"' % sql)
return Command(" ".join(base), **kwargs)
else:
raise NameError("Unrecognized or unsupported Postgres user operation: %s" % op)
PGSQL_MAPPINGS = {
'pg.client': psql,
'pg.createdatabase': pg_create_database,
'pg.createdb': pg_create_database,
'pg.createuser': pg_create_user,
'pg.database': pg_create_database,
'pg.database_exists': pg_database_exists,
'pg.db': pg_create_database,
'pg.dropdatabase': pg_drop_database,
'pg.dropdb': pg_drop_database,
'pg.dropuser': pg_drop_user,
'pg.dump': pg_dump_database,
'pg.dumpdb': pg_dump_database,
'pg.exists': pg_database_exists,
'pg.user': pg_create_user,
'psql': psql,
'pgsql.create': pgsql_create,
'pgsql.drop': pgsql_drop,
'pgsql.dump': pgsql_dump,
'pgsql.exists': pgsql_exists,
'pgsql.sql': pgsql_exec,
'pgsql.user': pgsql_user,
}

8
scripttease/version.py
Unescape View File

@ -1,5 +1,5 @@
DATE = "2020-09-17"
VERSION = "6.4.4-d"
DATE = "2020-09-22"
VERSION = "6.5.0-d"
MAJOR = 6
MINOR = 4
PATCH = 4
MINOR = 5
PATCH = 0

15
tests/examples/kitchen_sink.ini
Unescape View File

@ -162,26 +162,27 @@ symlink: /var/www/domains
touch: /path/to/file.txt
[create a postgres user/role]
pg.user: example_app
pgsql.user: example_app
[create a postgres database]
pg.db: example_app
pgsql.create: example_app
owner: example_app
[determine whether a postgres database exists]
pg.database_exists: example_app
pgsql.exists: example_app
[export a postgres database]
pg.dump: testing
pgsql.dump: testing
[drop a postgres user/role]
pg.dropuser: testing
pgsql.user: testing
op: drop
[drop a postgres database]
pg.dropdb: testing
pgsql.drop: testing
[run an SQL command on a postgres database]
psql: "SELECT * FROM projects WHERE category = 'testing'"
pgsql.sql: "SELECT * FROM projects WHERE category = 'testing'"
database: example_app
owner: example_app

60
tests/test_library_overlays_pgsql.py
Unescape View File

@ -1,8 +1,9 @@
import pytest
from scripttease.library.overlays.pgsql import *
def test_pg_create_database():
c = pg_create_database("testing", admin_pass="secret", template="mytemplate")
def test_pgsql_create():
c = pgsql_create("testing", admin_pass="secret", template="mytemplate")
s = c.get_statement()
assert "createdb" in s
assert "export PGPASSWORD=" in s
@ -14,47 +15,52 @@ def test_pg_create_database():
assert "testing" in s
def test_pg_create_user():
c = pg_create_user("testing", password="secret")
s = c.get_statement()
assert "createuser" in s
assert "-DRS" in s
assert "testing" in s
assert "ALTER USER testing" in s
def test_pg_database_exists():
c = pg_database_exists("testing")
def test_pgsql_exists():
c = pgsql_exists("testing")
s = c.get_statement()
assert "psql" in s
assert "testing_db_exists" in s
assert "pgsql_db_exists" in s
def test_pg_drop_database():
c = pg_drop_database("testing")
def test_pgsql_drop():
c = pgsql_drop("testing")
s = c.get_statement()
assert "dropdb" in s
assert "testing" in s
def test_pg_drop_user():
c = pg_drop_user("testing")
s = c.get_statement()
assert "dropuser" in s
assert "testing" in s
def test_pg_dump_database():
c = pg_dump_database("testing")
def test_pgsql_dump():
c = pgsql_dump("testing")
s = c.get_statement()
assert "pg_dump" in s
assert "--column-inserts" in s
assert "--file=testing.sql" in s
def test_psql():
c = psql("SELECT * FROM projects", database="testing")
def test_pgsql_exec():
c = pgsql_exec("SELECT * FROM projects", database="testing")
s = c.get_statement()
assert "psql" in s
assert "--dbname=testing" in s
assert '-c "SELECT * FROM projects"' in s
def test_pgsql_user():
c = pgsql_user("testing", password="secret")
s = c.get_statement()
assert "createuser" in s
assert "-DRS" in s
assert "testing" in s
assert "ALTER USER testing" in s
c = pgsql_user("testing", op="drop")
s = c.get_statement()
assert "dropuser" in s
assert "testing" in s
c = pgsql_user("testing", op="exists")
s = c.get_statement()
assert "SELECT 1 FROM pgsql_roles" in s
with pytest.raises(NameError):
pgsql_user("testing", op="nonexistent")

Write Preview
Loading…
Cancel
Save