You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
100 lines
2.8 KiB
100 lines
2.8 KiB
import pytest
|
|
from scripttease.exceptions import InvalidInput
|
|
from scripttease.lib.commands.pgsql import *
|
|
|
|
|
|
def test_pgsql_create():
|
|
c = pgsql_create("testing", password="secret", owner="testing", template="mytemplate")
|
|
s = c.get_statement()
|
|
assert "createdb" in s
|
|
assert "export PGPASSWORD=" in s
|
|
assert "--host=" in s
|
|
assert "--port=" in s
|
|
assert "-U" in s
|
|
assert "--owner=" in s
|
|
assert '--template="mytemplate"' in s
|
|
assert "testing" in s
|
|
|
|
|
|
def test_pgsql_exists():
|
|
c = pgsql_exists("testing")
|
|
s = c.get_statement()
|
|
assert "psql" in s
|
|
assert "testing_exists" in s
|
|
|
|
|
|
def test_pgsql_grant():
|
|
with pytest.raises(InvalidInput):
|
|
pgsql_grant("bob")
|
|
|
|
with pytest.raises(InvalidInput):
|
|
pgsql_grant("bob", database="testing")
|
|
|
|
c = pgsql_grant("bob", database="testing", schema="public")
|
|
s = c.get_statement()
|
|
assert '--dbname="testing"' in s
|
|
assert "GRANT ALL PRIVILEGES ON SCHEMA public TO bob" in s
|
|
|
|
c = pgsql_grant("bob", database="testing", table="testing")
|
|
s = c.get_statement()
|
|
assert "GRANT ALL PRIVILEGES ON TABLE testing TO bob" in s
|
|
|
|
|
|
def test_pgsql_drop():
|
|
c = pgsql_drop("testing")
|
|
s = c.get_statement()
|
|
assert "dropdb" in s
|
|
assert "testing" in s
|
|
|
|
|
|
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_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_load():
|
|
c = pgsql_load("testing", "path/to/file.sql", bogus=1)
|
|
s = c.get_statement()
|
|
assert '--dbname="testing"' in s
|
|
assert '--file="path/to/file.sql"' in s
|
|
assert '--bogus=1' in s # to test passing any key
|
|
|
|
|
|
def test_pgsql_sql():
|
|
c = pgsql_sql("UPDATE etl_testing SET is_processed = 't'", database="example_app")
|
|
s = c.get_statement()
|
|
# psql -U postgres --host=localhost --port=5432 --dbname="example_app" -c "UPDATE etl_testing SET is_processed = 't'"
|
|
assert '--dbname="example_app"' in s
|
|
assert '-c "UPDATE etl_testing SET is_processed = \'t\'"' 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(InvalidInput):
|
|
pgsql_user("testing", op="nonexistent")
|
|
|