# -*- coding: utf-8 -*- """Test rules""" from pyrseas.testutils import DatabaseToMapTestCase from pyrseas.testutils import InputMapToSqlTestCase, fix_indent CREATE_TABLE_STMT = "CREATE TABLE sd.t1 (c1 integer, c2 text)" CREATE_STMT = "CREATE RULE r1 AS ON %s TO sd.t1 DO %s" COMMENT_STMT = "COMMENT ON RULE r1 ON sd.t1 IS 'Test rule r1'" class RuleToMapTestCase(DatabaseToMapTestCase): """Test mapping of existing rules""" def test_map_rule_nothing(self): "Map a rule to do nothing" stmts = [CREATE_TABLE_STMT, CREATE_STMT % ('INSERT', 'NOTHING')] dbmap = self.to_map(stmts) assert dbmap['schema sd']['table t1']['rules'] == \ {'r1': {'event': 'insert', 'actions': 'NOTHING'}} def test_map_rule_instead(self): "Map rule with an INSTEAD action" stmts = [CREATE_TABLE_STMT, CREATE_STMT % ('UPDATE', 'INSTEAD NOTHING')] dbmap = self.to_map(stmts) assert dbmap['schema sd']['table t1']['rules'] == \ {'r1': {'event': 'update', 'instead': True, 'actions': 'NOTHING'}} def test_map_rule_conditional(self): "Map rule with a qualification" stmts = [CREATE_TABLE_STMT, "CREATE RULE r1 AS ON DELETE TO t1 WHERE OLD.c1 < 1000 " "DO INSERT INTO t1 VALUES (OLD.c1 + 1000, OLD.c2)"] dbmap = self.to_map(stmts) fmt = " %s\n %s" action = fmt % ("INSERT INTO sd.t1 (c1, c2)", " VALUES ((old.c1 + 1000), old.c2)") expmap = {'r1': {'event': 'delete', 'condition': "(old.c1 < 1000)", 'actions': action}} assert dbmap['schema sd']['table t1']['rules'] == expmap def test_map_rule_multi_actions(self): "Map rule with multiple actions" stmts = [CREATE_TABLE_STMT, CREATE_STMT % ( 'UPDATE', "(INSERT INTO t1 (c1) VALUES (old.c1 + 100); " "INSERT INTO t1 (c1) VALUES (old.c1 + 200))")] dbmap = self.to_map(stmts) ins = "INSERT INTO sd.t1 (c1)" fmt = "( %s\n %s\n %s\n %s\n)" actions = fmt % (ins, " VALUES ((old.c1 + 100));", ins, " VALUES ((old.c1 + 200));") expmap = {'r1': {'event': 'update', 'actions': actions}} assert dbmap['schema sd']['table t1']['rules'] == expmap def test_map_rule_comment(self): "Map a rule comment" stmts = [CREATE_TABLE_STMT, CREATE_STMT % ('INSERT', 'NOTHING'), COMMENT_STMT] dbmap = self.to_map(stmts) assert dbmap['schema sd']['table t1']['rules']['r1'][ 'description'] == 'Test rule r1' class RuleToSqlTestCase(InputMapToSqlTestCase): """Test SQL generation from input rules""" def test_create_rule_nothing(self): "Create a rule" inmap = self.std_map() inmap['schema sd'].update({'table t1': { 'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}], 'rules': {'r1': {'event': 'insert', 'actions': 'NOTHING'}}}}) sql = self.to_sql(inmap) assert fix_indent(sql[0]) == CREATE_TABLE_STMT assert fix_indent(sql[1]) == CREATE_STMT % ('INSERT', 'NOTHING') def test_create_rule_instead(self): "Create a rule with an INSTEAD action" inmap = self.std_map() inmap['schema sd'].update({'table t1': { 'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}], 'rules': {'r1': {'event': 'update', 'instead': True, 'actions': 'NOTHING'}}}}) sql = self.to_sql(inmap) assert fix_indent(sql[0]) == CREATE_TABLE_STMT assert fix_indent(sql[1]) == "CREATE RULE r1 AS ON UPDATE TO sd.t1 " \ "DO INSTEAD NOTHING" def test_create_rule_conditional(self): "Create a rule with qualification" inmap = self.std_map() inmap['schema sd'].update({'table t1': { 'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}], 'rules': {'r1': {'event': 'delete', 'condition': "old.c1 < 1000", 'actions': "INSERT INTO t1 VALUES (" "old.c1 + 1000, old.c2)"}}}}) sql = self.to_sql(inmap) assert fix_indent(sql[0]) == CREATE_TABLE_STMT assert fix_indent(sql[1]) == "CREATE RULE r1 AS ON DELETE TO sd.t1 " \ "WHERE old.c1 < 1000 DO INSERT INTO t1 VALUES " \ "(old.c1 + 1000, old.c2)" def test_create_rule_multi_actions(self): "Create a rule with multiple actions" inmap = self.std_map() inmap['schema sd'].update({'table t1': { 'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}], 'rules': {'r1': {'event': 'update', 'actions': "(INSERT INTO t1 VALUES (old.c1 + 100); " "INSERT INTO t1 VALUES (old.c1 + 200));)"}}}}) sql = self.to_sql(inmap) assert fix_indent(sql[0]) == CREATE_TABLE_STMT assert fix_indent(sql[1]) == "CREATE RULE r1 AS ON UPDATE " \ "TO sd.t1 DO (INSERT INTO t1 VALUES (old.c1 + 100); " \ "INSERT INTO t1 VALUES (old.c1 + 200));)" def test_create_rule_in_schema(self): "Create a rule within a non-default schema" inmap = self.std_map() inmap.update({'schema s1': { 'table t1': {'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}], 'rules': {'r1': {'event': 'insert', 'actions': 'NOTHING'}}}}}) sql = self.to_sql(inmap, ["CREATE SCHEMA s1"]) assert fix_indent(sql[1]) == "CREATE RULE r1 AS ON INSERT TO s1.t1 " \ "DO NOTHING" def test_drop_rule(self): "Drop an existing rule" stmts = [CREATE_TABLE_STMT, CREATE_STMT % ('INSERT', 'NOTHING')] inmap = self.std_map() inmap['schema sd'].update({'table t1': { 'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}]}}) sql = self.to_sql(inmap, stmts) assert sql == ["DROP RULE r1 ON sd.t1"] def test_drop_rule_table(self): "Drop an existing rule and the related table" stmts = [CREATE_TABLE_STMT, CREATE_STMT % ('INSERT', 'NOTHING')] sql = self.to_sql(self.std_map(), stmts) assert sql[0] == "DROP RULE r1 ON sd.t1" assert sql[1] == "DROP TABLE sd.t1" def test_rule_with_comment(self): "Create a rule with a comment" inmap = self.std_map() inmap['schema sd'].update({'table t1': { 'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}], 'rules': {'r1': {'event': 'insert', 'description': 'Test rule r1', 'actions': 'NOTHING'}}}}) sql = self.to_sql(inmap) assert sql[2] == COMMENT_STMT def test_comment_on_rule(self): "Create a comment on an existing rule" stmts = [CREATE_TABLE_STMT, CREATE_STMT % ('INSERT', 'NOTHING')] inmap = self.std_map() inmap['schema sd'].update({'table t1': { 'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}], 'rules': {'r1': {'event': 'insert', 'description': 'Test rule r1', 'actions': 'NOTHING'}}}}) sql = self.to_sql(inmap, stmts) assert sql == [COMMENT_STMT] def test_drop_rule_comment(self): "Drop the comment on an existing rule" stmts = [CREATE_TABLE_STMT, CREATE_STMT % ('INSERT', 'NOTHING'), COMMENT_STMT] inmap = self.std_map() inmap['schema sd'].update({'table t1': { 'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}], 'rules': {'r1': {'event': 'insert', 'actions': 'NOTHING'}}}}) sql = self.to_sql(inmap, stmts) assert sql == ["COMMENT ON RULE r1 ON sd.t1 IS NULL"] def test_change_rule_comment(self): "Change existing comment on a rule" stmts = [CREATE_TABLE_STMT, CREATE_STMT % ('INSERT', 'NOTHING'), COMMENT_STMT] inmap = self.std_map() inmap['schema sd'].update({'table t1': { 'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}], 'rules': {'r1': {'event': 'insert', 'description': 'Changed rule r1', 'actions': 'NOTHING'}}}}) sql = self.to_sql(inmap, stmts) assert sql == ["COMMENT ON RULE r1 ON sd.t1 IS 'Changed rule r1'"]