447 lines
21 KiB
Python
447 lines
21 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Test object privileges
|
|
|
|
The majority of other tests exclude access privileges. These
|
|
explicitly request it. In addition, the roles 'user1' and 'user2'
|
|
are created if they don't exist.
|
|
"""
|
|
|
|
from pyrseas.testutils import DatabaseToMapTestCase
|
|
from pyrseas.testutils import InputMapToSqlTestCase
|
|
|
|
CREATE_TABLE = "CREATE TABLE t1 (c1 integer, c2 text)"
|
|
SOURCE1 = "SELECT 'dummy'::text"
|
|
CREATE_FUNC = "CREATE FUNCTION f1() RETURNS text LANGUAGE sql IMMUTABLE AS " \
|
|
"$_$%s$_$" % SOURCE1
|
|
CREATE_FDW = "CREATE FOREIGN DATA WRAPPER fdw1"
|
|
CREATE_FS = "CREATE SERVER fs1 FOREIGN DATA WRAPPER fdw1"
|
|
GRANT_SELECT = "GRANT SELECT ON TABLE sd.t1 TO %s"
|
|
GRANT_INSUPD = "GRANT INSERT, UPDATE ON TABLE sd.t1 TO %s"
|
|
|
|
|
|
def check_extra_users(db):
|
|
"Check existence of extra test users"
|
|
for user in ['user1', 'user2']:
|
|
row = db.fetchone("SELECT 1 FROM pg_roles WHERE rolname = %s", (user,))
|
|
if row is None:
|
|
db.execute_commit("CREATE ROLE %s" % user)
|
|
|
|
|
|
class PrivilegeToMapTestCase(DatabaseToMapTestCase):
|
|
"""Test mapping of object privilege information"""
|
|
|
|
def setUp(self):
|
|
super(DatabaseToMapTestCase, self).setUp()
|
|
check_extra_users(self.db)
|
|
|
|
def test_map_schema(self):
|
|
"Map a schema with some GRANTs"
|
|
stmts = ["CREATE SCHEMA s1", "GRANT USAGE ON SCHEMA s1 TO PUBLIC",
|
|
"GRANT CREATE, USAGE ON SCHEMA s1 TO user1"]
|
|
dbmap = self.to_map(stmts, no_privs=False)
|
|
expmap = self.sort_privileges({'privileges': [
|
|
{self.db.user: ['all']}, {'PUBLIC': ['usage']},
|
|
{'user1': ['all']}]})
|
|
assert dbmap['schema s1'] == expmap
|
|
|
|
def test_map_table(self):
|
|
"Map a table with various GRANTs"
|
|
stmts = [CREATE_TABLE, GRANT_SELECT % 'PUBLIC', GRANT_INSUPD % 'user1',
|
|
"GRANT REFERENCES, TRIGGER ON t1 TO user2 WITH GRANT OPTION"]
|
|
dbmap = self.to_map(stmts, no_privs=False)
|
|
expmap = self.sort_privileges({'columns': [
|
|
{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'privileges': [{self.db.user: ['all']},
|
|
{'PUBLIC': ['select']},
|
|
{'user1': ['insert', 'update']},
|
|
{'user2': [{'trigger': {'grantable': True}},
|
|
{'references': {
|
|
'grantable': True}}]}]})
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_map_column(self):
|
|
"Map a table with GRANTs on column"
|
|
self.maxDiff = None
|
|
stmts = [CREATE_TABLE, GRANT_SELECT % 'PUBLIC',
|
|
"GRANT INSERT (c1, c2) ON t1 TO user1",
|
|
"GRANT INSERT (c2), UPDATE (c2) ON t1 TO user2"]
|
|
dbmap = self.to_map(stmts, no_privs=False)
|
|
expmap = self.sort_privileges({'columns': [
|
|
{'c1': {'type': 'integer', 'privileges': [{'user1': ['insert']}]}},
|
|
{'c2': {'type': 'text', 'privileges': [
|
|
{'user1': ['insert']}, {'user2': [
|
|
'insert', 'update']}]}}], 'privileges': [
|
|
{self.db.user: ['all']}, {'PUBLIC': ['select']}]})
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_map_sequence(self):
|
|
"Map a sequence with various GRANTs"
|
|
stmts = ["CREATE SEQUENCE seq1",
|
|
"GRANT SELECT ON SEQUENCE seq1 TO PUBLIC",
|
|
"GRANT USAGE, UPDATE ON SEQUENCE seq1 TO user1"]
|
|
dbmap = self.to_map(stmts, no_privs=False)
|
|
expmap = self.sort_privileges({'start_value': 1, 'increment_by': 1,
|
|
'max_value': None, 'min_value': None,
|
|
'cache_value': 1, 'privileges': [
|
|
{self.db.user: ['all']},
|
|
{'PUBLIC': ['select']},
|
|
{'user1': ['usage', 'update']}]})
|
|
assert dbmap['schema sd']['sequence seq1'] == expmap
|
|
|
|
def test_map_view(self):
|
|
"Map a view with various GRANTs"
|
|
stmts = ["CREATE VIEW v1 AS SELECT now()::date AS today",
|
|
"GRANT SELECT ON v1 TO PUBLIC",
|
|
"GRANT REFERENCES ON v1 TO user1"]
|
|
dbmap = self.to_map(stmts, no_privs=False)
|
|
expmap = self.sort_privileges(
|
|
{'columns': [{'today': {'type': 'date'}}],
|
|
'definition': " SELECT now()::date AS today;",
|
|
'privileges': [{self.db.user: ['all']},
|
|
{'PUBLIC': ['select']},
|
|
{'user1': ['references']}]})
|
|
assert dbmap['schema sd']['view v1'] == expmap
|
|
|
|
def test_map_function(self):
|
|
"Map a function with a GRANT and REVOKE from PUBLIC"
|
|
stmts = [CREATE_FUNC, "REVOKE ALL ON FUNCTION f1() FROM PUBLIC",
|
|
"GRANT EXECUTE ON FUNCTION f1() TO user1"]
|
|
dbmap = self.to_map(stmts, no_privs=False)
|
|
expmap = {'language': 'sql', 'returns': 'text',
|
|
'source': SOURCE1, 'volatility': 'immutable',
|
|
'privileges': [{self.db.user: ['execute']},
|
|
{'user1': ['execute']}]}
|
|
assert dbmap['schema sd']['function f1()'] == expmap
|
|
|
|
def test_map_fd_wrapper(self):
|
|
"Map a foreign data wrapper with a GRANT"
|
|
stmts = [CREATE_FDW,
|
|
"GRANT USAGE ON FOREIGN DATA WRAPPER fdw1 TO PUBLIC"]
|
|
dbmap = self.to_map(stmts, no_privs=False, superuser=True)
|
|
expmap = self.sort_privileges({'privileges':
|
|
[{self.db.user: ['usage']},
|
|
{'PUBLIC': ['usage']}]})
|
|
assert dbmap['foreign data wrapper fdw1'] == expmap
|
|
|
|
def test_map_server(self):
|
|
"Map a foreign server with a GRANT"
|
|
stmts = [CREATE_FDW, CREATE_FS,
|
|
"GRANT USAGE ON FOREIGN SERVER fs1 TO user1"]
|
|
dbmap = self.to_map(stmts, no_privs=False, superuser=True)
|
|
expmap = {'privileges': [{self.db.user: ['usage']},
|
|
{'user1': ['usage']}]}
|
|
assert dbmap['foreign data wrapper fdw1']['server fs1'] == expmap
|
|
|
|
def test_map_foreign_table(self):
|
|
"Map a foreign table with various GRANTs"
|
|
stmts = [CREATE_FDW, CREATE_FS,
|
|
"CREATE FOREIGN TABLE ft1 (c1 integer, c2 text) SERVER fs1",
|
|
"GRANT SELECT ON ft1 TO PUBLIC",
|
|
"GRANT INSERT, UPDATE ON ft1 TO user1"]
|
|
dbmap = self.to_map(stmts, no_privs=False, superuser=True)
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}], 'server': 'fs1',
|
|
'privileges': [{self.db.user: ['all']},
|
|
{'PUBLIC': ['select']},
|
|
{'user1': ['insert', 'update']}]}
|
|
assert dbmap['schema sd']['foreign table ft1'] == \
|
|
self.sort_privileges(expmap)
|
|
|
|
|
|
class PrivilegeToSqlTestCase(InputMapToSqlTestCase):
|
|
"""Test SQL generation of privilege information (GRANTs)"""
|
|
|
|
def setUp(self):
|
|
super(InputMapToSqlTestCase, self).setUp()
|
|
check_extra_users(self.db)
|
|
|
|
def test_create_schema(self):
|
|
"Create a schema with various privileges"
|
|
inmap = self.std_map()
|
|
inmap.update({'schema s1': {
|
|
'owner': self.db.user, 'privileges': [{
|
|
self.db.user: ['all']}, {'PUBLIC': ['usage', 'create']}]}})
|
|
sql = self.to_sql(inmap)
|
|
# sql[0] = CREATE SCHEMA
|
|
# sql[1] = ALTER SCHEMA OWNER
|
|
assert sql[2] == "GRANT ALL ON SCHEMA s1 TO %s" % self.db.user
|
|
assert sql[3] == "GRANT ALL ON SCHEMA s1 TO PUBLIC"
|
|
|
|
def test_schema_new_grant(self):
|
|
"Grant privileges on an existing schema"
|
|
inmap = self.std_map()
|
|
inmap.update({'schema s1': {
|
|
'owner': self.db.user, 'privileges': [{self.db.user: ['all']},
|
|
{'PUBLIC': ['create']}]}})
|
|
sql = sorted(self.to_sql(inmap, ["CREATE SCHEMA s1"]))
|
|
assert len(sql) == 2
|
|
assert sql[0] == "GRANT ALL ON SCHEMA s1 TO %s" % self.db.user
|
|
assert sql[1] == "GRANT CREATE ON SCHEMA s1 TO PUBLIC"
|
|
|
|
def test_create_table(self):
|
|
"Create a table with various privileges"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'PUBLIC': ['select']},
|
|
{'user1': ['insert', 'update']},
|
|
{'user2': [{'trigger': {'grantable': True}},
|
|
{'references': {'grantable': True}}]}]}})
|
|
sql = self.to_sql(inmap)
|
|
# sql[0] = CREATE TABLE
|
|
# sql[1] = ALTER TABLE OWNER
|
|
assert sql[2] == "GRANT ALL ON TABLE sd.t1 TO %s" % self.db.user
|
|
assert sql[3] == GRANT_SELECT % 'PUBLIC'
|
|
assert sql[4] == GRANT_INSUPD % 'user1'
|
|
assert sql[5] == "GRANT TRIGGER, REFERENCES ON TABLE sd.t1 " \
|
|
"TO user2 WITH GRANT OPTION"
|
|
|
|
def test_create_column_grants(self):
|
|
"Create a table with colum-level privileges"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer', 'privileges': [{'user1': [
|
|
'insert']}]}}, {'c2': {'type': 'text', 'privileges': [
|
|
{'user1': ['insert']}, {'user2': ['insert', 'update']}]}}],
|
|
'owner': self.db.user, 'privileges': [{self.db.user: ['all']},
|
|
{'PUBLIC': ['select']}]}})
|
|
sql = self.to_sql(inmap)
|
|
assert len(sql) == 7
|
|
# sql[0] = CREATE TABLE
|
|
# sql[1] = ALTER TABLE OWNER
|
|
assert sql[2] == "GRANT ALL ON TABLE sd.t1 TO %s" % self.db.user
|
|
assert sql[3] == GRANT_SELECT % 'PUBLIC'
|
|
assert sql[4] == "GRANT INSERT (c1) ON TABLE sd.t1 TO user1"
|
|
assert sql[5] == "GRANT INSERT (c2) ON TABLE sd.t1 TO user1"
|
|
assert sql[6] == "GRANT INSERT (c2), UPDATE (c2) ON TABLE sd.t1 " \
|
|
"TO user2"
|
|
|
|
def test_table_new_grant(self):
|
|
"Grant select privileges on an existing table"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'user1': ['select']}]}})
|
|
sql = self.to_sql(inmap, [CREATE_TABLE])
|
|
assert len(sql) == 2
|
|
sql = sorted(sql)
|
|
assert sql[0] == "GRANT ALL ON TABLE sd.t1 TO %s" % self.db.user
|
|
assert sql[1] == GRANT_SELECT % 'user1'
|
|
|
|
def test_table_change_grant(self):
|
|
"Grant select privileges on an existing table"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'PUBLIC': ['select']},
|
|
{'user1': ['insert', 'update']}]}})
|
|
sql = self.to_sql(inmap, [CREATE_TABLE, GRANT_SELECT % 'user1'])
|
|
assert len(sql) == 3
|
|
assert sorted(sql) == [GRANT_INSUPD % 'user1', GRANT_SELECT % 'PUBLIC',
|
|
"REVOKE SELECT ON TABLE sd.t1 FROM user1"]
|
|
|
|
def test_column_change_grants(self):
|
|
"Change existing colum-level privileges"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update(
|
|
{'table t1': {'columns': [{'c1': {
|
|
'type': 'integer', 'privileges': [
|
|
{'user1': ['insert']}, {'user2': ['insert', 'update']}]}},
|
|
{'c2': {'type': 'text', 'privileges': [
|
|
{'user1': ['insert']}]}}],
|
|
'owner': self.db.user, 'privileges': [
|
|
{self.db.user: ['all']}, {'PUBLIC': ['select']}]}})
|
|
stmts = [CREATE_TABLE, GRANT_SELECT % 'PUBLIC',
|
|
"GRANT INSERT (c1, c2) ON t1 TO user1",
|
|
"GRANT INSERT (c2), UPDATE (c2) ON t1 TO user2"]
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert len(sql) == 2
|
|
assert sql[0] == "GRANT INSERT (c1), UPDATE (c1) ON TABLE sd.t1 " \
|
|
"TO user2"
|
|
assert sql[1] == "REVOKE INSERT (c2), UPDATE (c2) ON TABLE sd.t1 " \
|
|
"FROM user2"
|
|
|
|
def test_table_revoke_all(self):
|
|
"Revoke all privileges on an existing table"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'owner': self.db.user}})
|
|
stmts = [CREATE_TABLE, GRANT_SELECT % 'PUBLIC', GRANT_INSUPD % 'user1']
|
|
sql = sorted(self.to_sql(inmap, stmts))
|
|
assert len(sql) == 3
|
|
assert sql[0] == "REVOKE ALL ON TABLE sd.t1 FROM %s" % self.db.user
|
|
assert sql[1] == "REVOKE INSERT, UPDATE ON TABLE sd.t1 FROM user1"
|
|
assert sql[2] == "REVOKE SELECT ON TABLE sd.t1 FROM PUBLIC"
|
|
|
|
def test_create_sequence(self):
|
|
"Create a sequence with some privileges"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'sequence seq1': {
|
|
'start_value': 1, 'increment_by': 1, 'max_value': None,
|
|
'min_value': None, 'cache_value': 1, 'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'PUBLIC': ['select']}]}})
|
|
sql = self.to_sql(inmap)
|
|
# sql[0] = CREATE SEQUENCE
|
|
# sql[1] = ALTER SEQUENCE OWNER
|
|
assert sql[2] == "GRANT ALL ON SEQUENCE sd.seq1 TO %s" % self.db.user
|
|
assert sql[3] == "GRANT SELECT ON SEQUENCE sd.seq1 TO PUBLIC"
|
|
|
|
def test_sequence_new_grant(self):
|
|
"Grant privileges on an existing sequence"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'sequence seq1': {
|
|
'start_value': 1, 'increment_by': 1, 'max_value': None,
|
|
'min_value': None, 'cache_value': 1, 'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'PUBLIC': ['select']}]}})
|
|
sql = sorted(self.to_sql(inmap, ["CREATE SEQUENCE seq1"]))
|
|
assert len(sql) == 2
|
|
assert sql[0] == "GRANT ALL ON SEQUENCE sd.seq1 TO %s" % self.db.user
|
|
assert sql[1] == "GRANT SELECT ON SEQUENCE sd.seq1 TO PUBLIC"
|
|
|
|
def test_create_view(self):
|
|
"Create a view with some privileges"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v1': {
|
|
'definition': " SELECT now()::date AS today;",
|
|
'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'user1': ['select']}]}})
|
|
sql = self.to_sql(inmap)
|
|
# sql[0] = CREATE VIEW
|
|
# sql[1] = ALTER VIEW OWNER
|
|
assert sql[2] == "GRANT ALL ON TABLE sd.v1 TO %s" % self.db.user
|
|
assert sql[3] == "GRANT SELECT ON TABLE sd.v1 TO user1"
|
|
|
|
def test_view_new_grant(self):
|
|
"Grant privileges on an existing view"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v1': {
|
|
'columns': [{'today': {'type': 'date'}}],
|
|
'definition': " SELECT now()::date AS today;",
|
|
'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'user1': ['select']}]}})
|
|
sql = sorted(self.to_sql(inmap, ["CREATE VIEW v1 AS "
|
|
"SELECT now()::date AS today"]))
|
|
assert len(sql) == 2
|
|
assert sql[0] == "GRANT ALL ON TABLE sd.v1 TO %s" % self.db.user
|
|
assert sql[1] == "GRANT SELECT ON TABLE sd.v1 TO user1"
|
|
|
|
def test_create_function(self):
|
|
"Create a function with some privileges"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'function f1()': {
|
|
'language': 'sql', 'returns': 'text', 'source': SOURCE1,
|
|
'volatility': 'immutable', 'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'PUBLIC': ['execute']}]}})
|
|
sql = self.to_sql(inmap)
|
|
# sql[0:1] = SET, CREATE FUNCTION
|
|
# sql[2] = ALTER FUNCTION OWNER
|
|
assert sql[3] == "GRANT EXECUTE ON FUNCTION sd.f1() TO %s" % (
|
|
self.db.user)
|
|
assert sql[4] == "GRANT EXECUTE ON FUNCTION sd.f1() TO PUBLIC"
|
|
|
|
def test_function_new_grant(self):
|
|
"Grant privileges on an existing function"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'function f1()': {
|
|
'language': 'sql', 'returns': 'text', 'source': SOURCE1,
|
|
'volatility': 'immutable', 'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'PUBLIC': ['execute']}]}})
|
|
sql = self.to_sql(inmap, [CREATE_FUNC])
|
|
assert len(sql) == 2
|
|
sql = sorted(sql)
|
|
# assumes self.db.user > PUBLIC
|
|
assert sql[0] == "GRANT EXECUTE ON FUNCTION sd.f1() TO PUBLIC"
|
|
assert sql[1] == "GRANT EXECUTE ON FUNCTION sd.f1() TO %s" % (
|
|
self.db.user)
|
|
|
|
def test_create_fd_wrapper(self):
|
|
"Create a foreign data wrapper with some privileges"
|
|
inmap = self.std_map()
|
|
inmap.update({'foreign data wrapper fdw1': {
|
|
'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'PUBLIC': ['usage']}]}})
|
|
sql = self.to_sql(inmap)
|
|
# sql[0] = CREATE FDW
|
|
# sql[1] = ALTER FDW OWNER
|
|
assert sql[2] == "GRANT USAGE ON FOREIGN DATA WRAPPER fdw1 " \
|
|
"TO %s" % self.db.user
|
|
assert sql[3] == "GRANT USAGE ON FOREIGN DATA WRAPPER fdw1 TO PUBLIC"
|
|
|
|
def test_fd_wrapper_new_grant(self):
|
|
"Grant privileges on an existing foreign data wrapper"
|
|
inmap = self.std_map()
|
|
inmap.update({'foreign data wrapper fdw1': {
|
|
'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'PUBLIC': ['usage']}]}})
|
|
sql = sorted(self.to_sql(inmap, [CREATE_FDW], superuser=True))
|
|
assert len(sql) == 2
|
|
# assumes self.db.user > PUBLIC
|
|
assert sql[0] == "GRANT USAGE ON FOREIGN DATA WRAPPER fdw1 TO PUBLIC"
|
|
assert sql[1] == "GRANT USAGE ON FOREIGN DATA WRAPPER fdw1 " \
|
|
"TO %s" % self.db.user
|
|
|
|
def test_create_server(self):
|
|
"Create a foreign server with some privileges"
|
|
inmap = self.std_map()
|
|
inmap.update({'foreign data wrapper fdw1': {'server fs1': {
|
|
'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'user2': ['usage']}]}}})
|
|
sql = self.to_sql(inmap, [CREATE_FDW], superuser=True)
|
|
# sql[0] = CREATE SERVER
|
|
# sql[1] = ALTER SERVER OWNER
|
|
assert sql[2] == "GRANT USAGE ON FOREIGN SERVER fs1 TO %s" % \
|
|
self.db.user
|
|
assert sql[3] == "GRANT USAGE ON FOREIGN SERVER fs1 TO user2"
|
|
|
|
def test_server_new_grant(self):
|
|
"Grant privileges on an existing foreign server"
|
|
inmap = self.std_map()
|
|
inmap.update({'foreign data wrapper fdw1': {'server fs1': {
|
|
'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'user2': ['usage']}]}}})
|
|
sql = sorted(self.to_sql(inmap, [CREATE_FDW, CREATE_FS],
|
|
superuser=True))
|
|
assert len(sql) == 2
|
|
assert sql[0] == "GRANT USAGE ON FOREIGN SERVER fs1 TO %s" % \
|
|
self.db.user
|
|
assert sql[1] == "GRANT USAGE ON FOREIGN SERVER fs1 TO user2"
|
|
|
|
def test_create_foreign_table(self):
|
|
"Create a foreign table with some privileges"
|
|
inmap = self.std_map()
|
|
inmap.update({'foreign data wrapper fdw1': {'server fs1': {}}})
|
|
inmap['schema sd'].update({'foreign table ft1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}], 'server': 'fs1',
|
|
'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'PUBLIC': ['select']},
|
|
{'user1': ['insert', 'update']}]}})
|
|
sql = self.to_sql(inmap, [CREATE_FDW, CREATE_FS], superuser=True)
|
|
# sql[0] = CREATE TABLE
|
|
# sql[1] = ALTER TABLE OWNER
|
|
assert sql[2] == "GRANT ALL ON TABLE sd.ft1 TO %s" % self.db.user
|
|
assert sql[3] == "GRANT SELECT ON TABLE sd.ft1 TO PUBLIC"
|
|
assert sql[4] == "GRANT INSERT, UPDATE ON TABLE sd.ft1 TO user1"
|
|
|
|
def test_foreign_table_new_grant(self):
|
|
"Grant privileges on an existing foreign table"
|
|
inmap = self.std_map()
|
|
inmap.update({'foreign data wrapper fdw1': {'server fs1': {}}})
|
|
inmap['schema sd'].update({'foreign table ft1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'server': 'fs1', 'owner': self.db.user,
|
|
'privileges': [{self.db.user: ['all']}, {'PUBLIC': ['select']},
|
|
{'user1': ['insert', 'update']}]}})
|
|
sql = sorted(self.to_sql(inmap, [
|
|
CREATE_FDW, CREATE_FS,
|
|
"CREATE FOREIGN TABLE ft1 (c1 integer, c2 text) SERVER fs1"],
|
|
superuser=True))
|
|
assert len(sql) == 3
|
|
assert sql[0] == "GRANT ALL ON TABLE sd.ft1 TO %s" % self.db.user
|
|
assert sql[1] == "GRANT INSERT, UPDATE ON TABLE sd.ft1 TO user1"
|
|
assert sql[2] == "GRANT SELECT ON TABLE sd.ft1 TO PUBLIC"
|