218 lines
8.9 KiB
Python
218 lines
8.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Test views"""
|
|
|
|
import pytest
|
|
|
|
from pyrseas.testutils import DatabaseToMapTestCase
|
|
from pyrseas.testutils import InputMapToSqlTestCase, fix_indent
|
|
|
|
CREATE_STMT = "CREATE VIEW sd.v1 AS SELECT now()::date AS today"
|
|
CREATE_TBL = "CREATE TABLE sd.t1 (c1 integer, c2 text, c3 integer)"
|
|
CREATE_STMT2 = "CREATE VIEW sd.v1 AS SELECT c1, c3 * 2 AS c2 FROM t1"
|
|
COMMENT_STMT = "COMMENT ON VIEW sd.v1 IS 'Test view v1'"
|
|
VIEW_DEFN = " SELECT now()::date AS today;"
|
|
|
|
|
|
class ViewToMapTestCase(DatabaseToMapTestCase):
|
|
"""Test mapping of created views"""
|
|
|
|
def test_map_view_no_table(self):
|
|
"Map a created view without a table dependency"
|
|
dbmap = self.to_map([CREATE_STMT])
|
|
expmap = {'columns': [{'today': {'type': 'date'}}],
|
|
'definition': VIEW_DEFN}
|
|
assert dbmap['schema sd']['view v1'] == expmap
|
|
|
|
def test_map_view_table(self):
|
|
"Map a created view with a table dependency"
|
|
stmts = [CREATE_TBL, CREATE_STMT2]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'integer'}}],
|
|
'depends_on': ['table t1'],
|
|
'definition': " SELECT t1.c1,"
|
|
"\n t1.c3 * 2 AS c2\n FROM sd.t1;"}
|
|
assert dbmap['schema sd']['view v1'] == expmap
|
|
|
|
def test_map_view_columns(self):
|
|
"Map a complex view's columns in addition to its definition"
|
|
stmts = ["CREATE TABLE t1 (c1 INTEGER UNIQUE)",
|
|
"CREATE TABLE t2 (c2 INTEGER PRIMARY KEY REFERENCES t1(c1))",
|
|
"CREATE VIEW v1 AS SELECT (ROW(t1.*)::t2).*, t1, 5 AS const "
|
|
"FROM t1"]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c2': {'type': 'integer'}},
|
|
{'t1': {'type': 'sd.t1'}},
|
|
{'const': {'type': 'integer'}}],
|
|
'definition': " SELECT (ROW(t1.c1)::sd.t2).c2 AS c2,"
|
|
"\n t1.*::sd.t1 AS t1,\n 5 AS const\n FROM sd.t1;",
|
|
'depends_on': ['table t1']}
|
|
assert dbmap['schema sd']['view v1'] == expmap
|
|
|
|
def test_map_view_comment(self):
|
|
"Map a view with a comment"
|
|
dbmap = self.to_map([CREATE_STMT, COMMENT_STMT])
|
|
assert dbmap['schema sd']['view v1']['description'] == \
|
|
'Test view v1'
|
|
|
|
|
|
class ViewToSqlTestCase(InputMapToSqlTestCase):
|
|
"""Test SQL generation from input views"""
|
|
|
|
def test_create_view_no_table(self):
|
|
"Create a view with no table dependency"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v1': {'definition': VIEW_DEFN}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == CREATE_STMT
|
|
|
|
def test_create_view(self):
|
|
"Create a view"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}},
|
|
{'c3': {'type': 'integer'}}]}})
|
|
inmap['schema sd'].update({'view v1': {
|
|
'definition': "SELECT c1, c3 * 2 AS c2 FROM t1"}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == CREATE_TBL
|
|
assert fix_indent(sql[1]) == CREATE_STMT2
|
|
|
|
def test_create_view_in_schema(self):
|
|
"Create a view within a non-default schema"
|
|
inmap = self.std_map()
|
|
inmap.update({'schema s1': {'view v1': {'definition': VIEW_DEFN}}})
|
|
sql = self.to_sql(inmap, ["CREATE SCHEMA s1"])
|
|
assert fix_indent(sql[0]) == \
|
|
"CREATE VIEW s1.v1 AS SELECT now()::date AS today"
|
|
|
|
def test_bad_view_map(self):
|
|
"Error creating a view with a bad map"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'v1': {'definition': VIEW_DEFN}})
|
|
with pytest.raises(KeyError):
|
|
self.to_sql(inmap)
|
|
|
|
def test_drop_view_no_table(self):
|
|
"Drop an existing view without a table dependency"
|
|
sql = self.to_sql(self.std_map(), [CREATE_STMT])
|
|
assert sql == ["DROP VIEW sd.v1"]
|
|
|
|
def test_drop_view(self):
|
|
"Drop an existing view with table dependencies"
|
|
stmts = ["CREATE TABLE t1 (c1 INTEGER, c2 TEXT)",
|
|
"CREATE TABLE t2 (c1 INTEGER, c3 TEXT)",
|
|
"CREATE VIEW v1 AS SELECT t1.c1, c2, c3 "
|
|
"FROM t1 JOIN t2 ON (t1.c1 = t2.c1)"]
|
|
sql = self.to_sql(self.std_map(), stmts)
|
|
assert sql[0] == "DROP VIEW sd.v1"
|
|
# can't control which table will be dropped first
|
|
drt1 = 1
|
|
drt2 = 2
|
|
if 't1' in sql[2]:
|
|
drt1 = 2
|
|
drt2 = 1
|
|
assert sql[drt1] == "DROP TABLE sd.t1"
|
|
assert sql[drt2] == "DROP TABLE sd.t2"
|
|
|
|
def test_rename_view(self):
|
|
"Rename an existing view"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v2': {
|
|
'oldname': 'v1', 'definition': VIEW_DEFN}})
|
|
sql = self.to_sql(inmap, [CREATE_STMT])
|
|
assert sql == ["ALTER VIEW sd.v1 RENAME TO v2"]
|
|
|
|
def test_bad_rename_view(self):
|
|
"Error renaming a non-existing view"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v2': {
|
|
'oldname': 'v3', 'definition': VIEW_DEFN}})
|
|
with pytest.raises(KeyError):
|
|
self.to_sql(inmap, [CREATE_STMT])
|
|
|
|
def test_change_view_defn(self):
|
|
"Change view definition"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v1': {
|
|
'columns': [{'today': {'type': 'date'}}],
|
|
'definition': " SELECT 'now'::text::date AS today;"}})
|
|
sql = self.to_sql(inmap, [CREATE_STMT])
|
|
assert fix_indent(sql[0]) == "CREATE OR REPLACE VIEW sd.v1 AS " \
|
|
"SELECT 'now'::text::date AS today"
|
|
|
|
def test_change_column_name(self):
|
|
"Attempt rename view column name (disallowed)"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v1': {
|
|
'columns': [{'todays_date': {'type': 'date'}}],
|
|
'definition': " SELECT now()::date AS todays_date;"}})
|
|
with pytest.raises(KeyError):
|
|
sql = self.to_sql(inmap, [CREATE_STMT])
|
|
|
|
def test_change_column_type(self):
|
|
"Change view column type to different type (disallowed)"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'numeric'}}],
|
|
'definition': " SELECT c1, c3 * 2.0 AS c2 FROM t1;"}})
|
|
with pytest.raises(TypeError):
|
|
sql = self.to_sql(inmap, [CREATE_TBL, CREATE_STMT2])
|
|
|
|
def test_view_depend_pk(self):
|
|
"Create a view that depends on a primary key. See issue #72"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'primary_key': {'t1_pkey': {'columns': ['c1']}}},
|
|
'view v1': {
|
|
'definition': " SELECT t1.c1,\n t1.c2\n FROM t1\n "
|
|
"GROUP BY t1.c1;",
|
|
'depends_on': ['table t1']}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == "CREATE TABLE sd.t1 (c1 integer, c2 text)"
|
|
assert fix_indent(sql[1]) == "ALTER TABLE sd.t1 ADD CONSTRAINT " \
|
|
"t1_pkey PRIMARY KEY (c1)"
|
|
assert fix_indent(sql[2]) == "CREATE VIEW sd.v1 AS SELECT t1.c1, " \
|
|
"t1.c2 FROM t1 GROUP BY t1.c1"
|
|
|
|
def test_view_with_comment(self):
|
|
"Create a view with a comment"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v1': {
|
|
'definition': VIEW_DEFN, 'description': "Test view v1"}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == CREATE_STMT
|
|
assert sql[1] == COMMENT_STMT
|
|
|
|
def test_comment_on_view(self):
|
|
"Create a comment for an existing view"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v1': {
|
|
'columns': [{'today': {'type': 'date'}}],
|
|
'definition': VIEW_DEFN, 'description': "Test view v1"}})
|
|
sql = self.to_sql(inmap, [CREATE_STMT])
|
|
assert sql == [COMMENT_STMT]
|
|
|
|
def test_drop_view_comment(self):
|
|
"Drop the comment on an existing view"
|
|
stmts = [CREATE_STMT, COMMENT_STMT]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v1': {
|
|
'columns': [{'today': {'type': 'date'}}],
|
|
'definition': VIEW_DEFN}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql == ["COMMENT ON VIEW sd.v1 IS NULL"]
|
|
|
|
def test_change_view_comment(self):
|
|
"Change existing comment on a view"
|
|
stmts = [CREATE_STMT, COMMENT_STMT]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'view v1': {
|
|
'columns': [{'today': {'type': 'date'}}],
|
|
'definition': VIEW_DEFN, 'description': "Changed view v1"}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql == ["COMMENT ON VIEW sd.v1 IS 'Changed view v1'"]
|