437 lines
19 KiB
Python
437 lines
19 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Test tables"""
|
|
|
|
import pytest
|
|
|
|
from pyrseas.testutils import DatabaseToMapTestCase
|
|
from pyrseas.testutils import InputMapToSqlTestCase, fix_indent
|
|
|
|
CREATE_STMT = "CREATE TABLE sd.t1 (c1 integer, c2 text)"
|
|
COMMENT_STMT = "COMMENT ON TABLE sd.t1 IS 'Test table t1'"
|
|
CREATE_STOR_PARAMS = CREATE_STMT + \
|
|
" WITH (fillfactor=90, autovacuum_enabled=false)"
|
|
|
|
|
|
class TableToMapTestCase(DatabaseToMapTestCase):
|
|
"""Test mapping of created tables"""
|
|
|
|
def test_map_table_simple(self):
|
|
"Map a table with two columns"
|
|
dbmap = self.to_map([CREATE_STMT])
|
|
assert dbmap['schema sd']['table t1'] == {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}]}
|
|
|
|
def test_map_table_comment(self):
|
|
"Map a table comment"
|
|
dbmap = self.to_map([CREATE_STMT, COMMENT_STMT])
|
|
assert dbmap['schema sd']['table t1'] == {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'description': 'Test table t1'}
|
|
|
|
def test_map_table_comment_quotes(self):
|
|
"Map a table comment with quotes"
|
|
stmts = [CREATE_STMT, "COMMENT ON TABLE t1 IS "
|
|
"'A \"special\" person''s table t1'"]
|
|
dbmap = self.to_map(stmts)
|
|
assert dbmap['schema sd']['table t1'] == {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'description': "A \"special\" person's table t1"}
|
|
|
|
def test_map_column_comments(self):
|
|
"Map two column comments"
|
|
stmts = [CREATE_STMT,
|
|
"COMMENT ON COLUMN t1.c1 IS 'Test column c1 of t1'",
|
|
"COMMENT ON COLUMN t1.c2 IS 'Test column c2 of t1'"]
|
|
dbmap = self.to_map(stmts)
|
|
assert dbmap['schema sd']['table t1'] == {
|
|
'columns': [{'c1': {'type': 'integer',
|
|
'description': 'Test column c1 of t1'}},
|
|
{'c2': {'type': 'text',
|
|
'description': 'Test column c2 of t1'}}]}
|
|
|
|
def test_map_table_options(self):
|
|
"Map a table with options"
|
|
dbmap = self.to_map([CREATE_STOR_PARAMS])
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'options': ["fillfactor=90", 'autovacuum_enabled=false']}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_map_inherit(self):
|
|
"Map a table that inherits from two other tables"
|
|
stmts = [CREATE_STMT, "CREATE TABLE t2 (c3 integer)",
|
|
"CREATE TABLE t3 (c4 text) INHERITS (t1, t2)"]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c1': {'type': 'integer', 'inherited': True}},
|
|
{'c2': {'type': 'text', 'inherited': True}},
|
|
{'c3': {'type': 'integer', 'inherited': True}},
|
|
{'c4': {'type': 'text'}}],
|
|
'inherits': ['t1', 't2']}
|
|
assert dbmap['schema sd']['table t3'] == expmap
|
|
|
|
def test_map_inherit_delim(self):
|
|
"Map a table that inherits from two other tables (delim identifiers)"
|
|
stmts = [CREATE_STMT, "CREATE TABLE \"t-2\" (c3 integer)",
|
|
"CREATE TABLE \"t-3\" (c4 text) INHERITS (t1, \"t-2\")"]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c1': {'type': 'integer', 'inherited': True}},
|
|
{'c2': {'type': 'text', 'inherited': True}},
|
|
{'c3': {'type': 'integer', 'inherited': True}},
|
|
{'c4': {'type': 'text'}}],
|
|
"inherits": ["t1", "t-2"]}
|
|
assert dbmap["schema sd"]["table t-3"] == expmap
|
|
|
|
def test_map_unlogged_table(self):
|
|
"Map an unlogged table"
|
|
dbmap = self.to_map(["CREATE UNLOGGED TABLE t1 (c1 integer, c2 text)"])
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'unlogged': True}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_map_table_within_schema(self):
|
|
"Map a schema and a table within it"
|
|
stmts = ["CREATE SCHEMA s1",
|
|
"CREATE TABLE s1.t1 (c1 INTEGER, c2 TEXT)"]
|
|
dbmap = self.to_map(stmts)
|
|
assert dbmap['schema s1'] == {
|
|
'table t1': {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]}}
|
|
|
|
def test_map_table_quoted(self):
|
|
"Map a schema and a table both of which need to be quoted"
|
|
stmts = ['CREATE SCHEMA "a schema"',
|
|
'CREATE TABLE "a schema"."The.Table" ("column" SERIAL, '
|
|
'c2 TEXT)']
|
|
dbmap = self.to_map(stmts)
|
|
assert dbmap['schema a schema']['table The.Table'] == {
|
|
'columns': [{'column': {
|
|
'type': 'integer', 'not_null': True,
|
|
'default':
|
|
'nextval(\'"a schema"."The.Table_column_seq"\'::regclass)'}},
|
|
{'c2': {'type': 'text'}}]}
|
|
|
|
def test_map_select_tables(self):
|
|
"Map two tables out of three present"
|
|
stmts = [CREATE_STMT, "CREATE TABLE t2 (c1 integer, c2 text)",
|
|
"CREATE TABLE t3 (c1 integer, c2 text)"]
|
|
dbmap = self.to_map(stmts, tables=['t2', 't1'])
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
assert dbmap['schema sd']['table t2'] == expmap
|
|
assert 'table t3' not in dbmap['schema sd']
|
|
|
|
def test_map_table_sequence(self):
|
|
"Map sequence if owned by a table"
|
|
stmts = [CREATE_STMT, "CREATE TABLE t2 (c1 integer, c2 text)",
|
|
"CREATE SEQUENCE seq1", "ALTER SEQUENCE seq1 OWNED BY t2.c1",
|
|
"CREATE SEQUENCE seq2"]
|
|
dbmap = self.to_map(stmts, tables=['t2'])
|
|
self.db.execute_commit("DROP SEQUENCE seq1")
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]}
|
|
assert 'table t1' not in dbmap['schema sd']
|
|
assert dbmap['schema sd']['table t2'] == expmap
|
|
assert 'sequence seq1' in dbmap['schema sd']
|
|
assert 'sequence seq2' not in dbmap['schema sd']
|
|
|
|
def test_map_partition_range(self):
|
|
"Map a partitioned table with two partitions by range"
|
|
spec1 = "FROM ('2015-01-01', MINVALUE) TO ('2016-12-31', 5)"
|
|
stmts = ["CREATE TABLE t1 (c1 date, c2 integer, c3 text) "
|
|
"PARTITION BY RANGE (c1, c2)",
|
|
"CREATE TABLE t1a PARTITION OF t1 FOR VALUES %s" % spec1,
|
|
"CREATE TABLE t1b PARTITION OF t1 FOR VALUES "
|
|
"FROM ('2017-01-01', 11) TO ('2020-12-31', 15)"]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c1': {'type': 'date'}},
|
|
{'c2': {'type': 'integer'}},
|
|
{'c3': {'type': 'text'}}],
|
|
'partition_by': {'range': ['c1', 'c2']}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
expmap2 = {'partition_bound_spec': spec1, 'partition_of': 't1'}
|
|
assert dbmap['schema sd']['table t1a'] == expmap2
|
|
|
|
def test_map_partition_list(self):
|
|
"Map a partitioned table with two partitions by list"
|
|
spec1 = "IN (1, 3, 5, 7)"
|
|
stmts = ["CREATE TABLE t1 (c1 integer, c2 text) "
|
|
"PARTITION BY LIST (c1)",
|
|
"CREATE TABLE t1a PARTITION OF t1 FOR VALUES %s" % spec1,
|
|
"CREATE TABLE t1b PARTITION OF t1 FOR VALUES IN (2, 4, 6)"]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'partition_by': {'list': ['c1']}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
expmap2 = {'partition_bound_spec': spec1, 'partition_of': 't1'}
|
|
assert dbmap['schema sd']['table t1a'] == expmap2
|
|
|
|
|
|
class TableToSqlTestCase(InputMapToSqlTestCase):
|
|
"""Test SQL generation of table statements from input schemas"""
|
|
|
|
def test_create_table_simple(self):
|
|
"Create a two-column table"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == CREATE_STMT
|
|
|
|
def test_create_table_quoted_idents(self):
|
|
"Create a table needing quoted identifiers"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table order': {
|
|
'columns': [{'primary': {'type': 'integer'}},
|
|
{'two words': {'type': 'text'}}]}})
|
|
sql = self.to_sql(inmap, quote_reserved=True)
|
|
assert fix_indent(sql[0]) == 'CREATE TABLE sd."order" (' \
|
|
'"primary" integer, "two words" text)'
|
|
|
|
def test_bad_table_map(self):
|
|
"Error creating a table with a bad map"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]}})
|
|
with pytest.raises(KeyError):
|
|
self.to_sql(inmap)
|
|
|
|
def test_missing_columns(self):
|
|
"Error creating a table with no columns"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {'columns': []}})
|
|
with pytest.raises(ValueError):
|
|
self.to_sql(inmap)
|
|
|
|
def test_drop_table(self):
|
|
"Drop an existing table"
|
|
sql = self.to_sql(self.std_map(), [CREATE_STMT])
|
|
assert sql == ["DROP TABLE sd.t1"]
|
|
|
|
def test_rename_table(self):
|
|
"Rename an existing table"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t2': {
|
|
'oldname': 't1',
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]}})
|
|
sql = self.to_sql(inmap, [CREATE_STMT])
|
|
assert sql == ["ALTER TABLE sd.t1 RENAME TO t2"]
|
|
|
|
def test_create_table_options(self):
|
|
"Create a table with options"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'options': ["fillfactor=90", "autovacuum_enabled=false"]}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == CREATE_STOR_PARAMS
|
|
|
|
def test_change_table_options(self):
|
|
"Change a table's storage parameters"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'options': ["fillfactor=70"]}})
|
|
sql = self.to_sql(inmap, [CREATE_STOR_PARAMS])
|
|
assert fix_indent(sql[0]) == "ALTER TABLE sd.t1 SET (fillfactor=70)," \
|
|
" RESET (autovacuum_enabled)"
|
|
|
|
def test_create_table_within_schema(self):
|
|
"Create a new schema and a table within it"
|
|
inmap = self.std_map()
|
|
inmap.update({'schema s1': {'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]}}})
|
|
sql = self.to_sql(inmap)
|
|
expsql = ["CREATE SCHEMA s1",
|
|
"CREATE TABLE s1.t1 (c1 integer, c2 text)"]
|
|
for i in range(len(expsql)):
|
|
assert fix_indent(sql[i]) == expsql[i]
|
|
|
|
def test_unlogged_table(self):
|
|
"Create an unlogged table"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}], 'unlogged': True}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == \
|
|
"CREATE UNLOGGED TABLE sd.t1 (c1 integer, c2 text)"
|
|
|
|
def test_table_owned_by_sequence(self):
|
|
"Alter a table to be owned by a table column"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]},
|
|
'sequence seq1': {
|
|
'cache_value': 1, 'increment_by': 1, 'max_value': None,
|
|
'min_value': None, 'start_value': 1,
|
|
'owner_table': 't1', 'owner_column': 'c1'}})
|
|
sql = self.to_sql(inmap, [CREATE_STMT, "CREATE SEQUENCE seq1"])
|
|
assert sql[0] == "ALTER SEQUENCE sd.seq1 OWNED BY sd.t1.c1"
|
|
|
|
def test_create_partitioned_tables(self):
|
|
"Create a partitioned table and two partitions"
|
|
inmap = self.std_map()
|
|
spec1 = "FROM ('2015-01-01', MINVALUE) TO ('2016-12-31', 5)"
|
|
spec2 = "FROM ('2017-01-01', 11) TO ('2020-12-31', 15)"
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'date'}}, {'c2': {'type': 'integer'}},
|
|
{'c3': {'type': 'text'}}],
|
|
'partition_by': {'range': ['c1', 'c2']}}, 'table t1a': {
|
|
'partition_bound_spec': spec1, 'partition_of': 't1'},
|
|
'table t1b': {'partition_bound_spec': spec2,
|
|
'partition_of': 't1'}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == "CREATE TABLE sd.t1 (c1 date, " \
|
|
"c2 integer, c3 text) PARTITION BY RANGE (c1, c2)"
|
|
assert fix_indent(sql[1]) == (
|
|
"CREATE TABLE sd.t1a PARTITION OF t1 FOR VALUES %s" % spec1)
|
|
assert fix_indent(sql[2]) == (
|
|
"CREATE TABLE sd.t1b PARTITION OF t1 FOR VALUES %s" % spec2)
|
|
|
|
|
|
class TableCommentToSqlTestCase(InputMapToSqlTestCase):
|
|
"""Test SQL generation of table and column COMMENT statements"""
|
|
|
|
def _tblmap(self):
|
|
"Return a table input map with a comment"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'description': 'Test table t1',
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]}})
|
|
return inmap
|
|
|
|
def test_table_with_comment(self):
|
|
"Create a table with a comment"
|
|
sql = self.to_sql(self._tblmap())
|
|
assert fix_indent(sql[0]) == CREATE_STMT
|
|
assert sql[1] == COMMENT_STMT
|
|
|
|
def test_comment_on_table(self):
|
|
"Create a comment for an existing table"
|
|
sql = self.to_sql(self._tblmap(), [CREATE_STMT])
|
|
assert sql == [COMMENT_STMT]
|
|
|
|
def test_table_comment_quotes(self):
|
|
"Create a table comment with quotes"
|
|
inmap = self._tblmap()
|
|
inmap['schema sd']['table t1']['description'] = \
|
|
"A \"special\" person's table t1"
|
|
sql = self.to_sql(inmap, [CREATE_STMT])
|
|
assert sql == ["COMMENT ON TABLE sd.t1 IS "
|
|
"'A \"special\" person''s table t1'"]
|
|
|
|
def test_drop_table_comment(self):
|
|
"Drop a comment on an existing table"
|
|
inmap = self._tblmap()
|
|
del inmap['schema sd']['table t1']['description']
|
|
sql = self.to_sql(inmap, [CREATE_STMT, COMMENT_STMT])
|
|
assert sql == ["COMMENT ON TABLE sd.t1 IS NULL"]
|
|
|
|
def test_change_table_comment(self):
|
|
"Change existing comment on a table"
|
|
inmap = self._tblmap()
|
|
inmap['schema sd']['table t1'].update(
|
|
{'description': 'Changed table t1'})
|
|
sql = self.to_sql(inmap, [CREATE_STMT, COMMENT_STMT])
|
|
assert sql == ["COMMENT ON TABLE sd.t1 IS 'Changed table t1'"]
|
|
|
|
def test_create_column_comments(self):
|
|
"Create a table with column comments"
|
|
inmap = self._tblmap()
|
|
inmap['schema sd']['table t1']['columns'][0]['c1'].update(
|
|
description='Test column c1')
|
|
inmap['schema sd']['table t1']['columns'][1]['c2'].update(
|
|
description='Test column c2')
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == CREATE_STMT
|
|
assert sql[1] == COMMENT_STMT
|
|
assert sql[2] == "COMMENT ON COLUMN sd.t1.c1 IS 'Test column c1'"
|
|
assert sql[3] == "COMMENT ON COLUMN sd.t1.c2 IS 'Test column c2'"
|
|
|
|
def test_add_column_comment(self):
|
|
"Add a column comment to an existing table"
|
|
inmap = self._tblmap()
|
|
inmap['schema sd']['table t1']['columns'][0]['c1'].update(
|
|
description='Test column c1')
|
|
sql = self.to_sql(inmap, [CREATE_STMT, COMMENT_STMT])
|
|
assert sql[0] == "COMMENT ON COLUMN sd.t1.c1 IS 'Test column c1'"
|
|
|
|
def test_add_column_with_comment(self):
|
|
"Add a commented column to an existing table"
|
|
inmap = self._tblmap()
|
|
inmap['schema sd']['table t1']['columns'].append({'c3': {
|
|
'description': 'Test column c3', 'type': 'integer'}})
|
|
sql = self.to_sql(inmap, [CREATE_STMT, COMMENT_STMT])
|
|
assert fix_indent(sql[0]) == "ALTER TABLE sd.t1 ADD COLUMN c3 integer"
|
|
assert sql[1] == "COMMENT ON COLUMN sd.t1.c3 IS 'Test column c3'"
|
|
|
|
def test_drop_column_comment(self):
|
|
"Drop a column comment on an existing table"
|
|
stmts = [CREATE_STMT, COMMENT_STMT,
|
|
"COMMENT ON COLUMN t1.c1 IS 'Test column c1'"]
|
|
sql = self.to_sql(self._tblmap(), stmts)
|
|
assert sql[0] == "COMMENT ON COLUMN sd.t1.c1 IS NULL"
|
|
|
|
def test_change_column_comment(self):
|
|
"Add a column comment to an existing table"
|
|
inmap = self._tblmap()
|
|
inmap['schema sd']['table t1']['columns'][0]['c1'].update(
|
|
description='Changed column c1')
|
|
sql = self.to_sql(inmap, [CREATE_STMT, COMMENT_STMT])
|
|
assert sql[0] == "COMMENT ON COLUMN sd.t1.c1 IS 'Changed column c1'"
|
|
|
|
|
|
class TableInheritToSqlTestCase(InputMapToSqlTestCase):
|
|
"""Test SQL generation of table inheritance statements"""
|
|
|
|
def test_table_inheritance(self):
|
|
"Create a table that inherits from another"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]}})
|
|
inmap['schema sd'].update({'table t2': {
|
|
'columns': [{'c1': {'type': 'integer', 'inherited': True}},
|
|
{'c2': {'type': 'text', 'inherited': True}},
|
|
{'c3': {'type': 'numeric'}}], 'inherits': ['t1']}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == CREATE_STMT
|
|
assert fix_indent(sql[1]) == "CREATE TABLE sd.t2 (c3 numeric) " \
|
|
"INHERITS (sd.t1)"
|
|
|
|
def test_table_inherit_delim(self):
|
|
"Create a table that inherits from another (delimited identifiers)"
|
|
inmap = self.std_map()
|
|
inmap.update({'schema s-d': {'table t-1': {
|
|
'columns': [{'c-1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}]}}})
|
|
inmap['schema s-d'].update({'table t-2': {
|
|
'columns': [{'c-1': {'type': 'integer', 'inherited': True}},
|
|
{'c2': {'type': 'text', 'inherited': True}},
|
|
{'c3': {'type': 'numeric'}}], 'inherits': ['t-1']}})
|
|
sql = self.to_sql(inmap, ["CREATE SCHEMA \"s-d\""])
|
|
print(sql)
|
|
assert fix_indent(sql[0]) == "CREATE TABLE \"s-d\".\"t-1\" " \
|
|
"(\"c-1\" integer, c2 text)"
|
|
assert fix_indent(sql[1]) == "CREATE TABLE \"s-d\".\"t-2\" " \
|
|
"(c3 numeric) INHERITS (\"s-d\".\"t-1\")"
|
|
|
|
def test_drop_inherited(self):
|
|
"Drop tables that inherit from others"
|
|
stmts = [CREATE_STMT, "CREATE TABLE t2 (c3 numeric) INHERITS (t1)",
|
|
"CREATE TABLE t3 (c4 date) INHERITS (t2)"]
|
|
sql = self.to_sql(self.std_map(), stmts)
|
|
assert sql == ["DROP TABLE sd.t3", "DROP TABLE sd.t2",
|
|
"DROP TABLE sd.t1"]
|