161 lines
7.3 KiB
Python
161 lines
7.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Test operator classes"""
|
|
|
|
import pytest
|
|
|
|
from pyrseas.testutils import DatabaseToMapTestCase
|
|
from pyrseas.testutils import InputMapToSqlTestCase, fix_indent
|
|
|
|
CREATE_TYPE_STMT = """
|
|
CREATE TYPE myint;
|
|
CREATE FUNCTION myintin(cstring) RETURNS myint AS 'int4in' LANGUAGE internal;
|
|
CREATE FUNCTION myintout(myint) RETURNS cstring AS 'int4out' LANGUAGE internal;
|
|
CREATE TYPE myint (INPUT = myintin, OUTPUT = myintout);
|
|
CREATE FUNCTION myinteq(myint,myint) RETURNS boolean AS 'int4eq'
|
|
LANGUAGE internal;
|
|
CREATE FUNCTION myintlt(myint,myint) RETURNS boolean AS 'int4lt'
|
|
LANGUAGE internal;
|
|
CREATE OPERATOR < (PROCEDURE = myintlt, LEFTARG = myint, RIGHTARG = myint);
|
|
CREATE OPERATOR = (PROCEDURE = myinteq, LEFTARG = myint, RIGHTARG = myint);
|
|
CREATE FUNCTION btmyintcmp(myint,myint) RETURNS integer AS 'btint4cmp'
|
|
LANGUAGE internal;
|
|
"""
|
|
|
|
CREATE_STMT = "CREATE OPERATOR CLASS oc1 FOR TYPE integer USING btree " \
|
|
"AS OPERATOR 1 <, OPERATOR 3 =, FUNCTION 1 btint4cmp(integer,integer)"
|
|
CREATE_STMT_LONG = "CREATE OPERATOR CLASS sd.oc1 FOR TYPE integer " \
|
|
"USING btree AS OPERATOR 1 <(integer,integer), " \
|
|
"OPERATOR 3 =(integer,integer), FUNCTION 1 btint4cmp(integer,integer)"
|
|
COMMENT_STMT = "COMMENT ON OPERATOR CLASS sd.oc1 USING btree IS " \
|
|
"'Test operator class oc1'"
|
|
|
|
|
|
class OperatorClassToMapTestCase(DatabaseToMapTestCase):
|
|
"""Test mapping of existing operator classes"""
|
|
|
|
superuser = True
|
|
|
|
def test_map_operclass1(self):
|
|
"Map an operator class"
|
|
dbmap = self.to_map([CREATE_STMT])
|
|
assert dbmap['schema sd']['operator class oc1 using btree'] == \
|
|
{'type': 'integer',
|
|
'operators': {1: '<(integer,integer)', 3: '=(integer,integer)'},
|
|
'functions': {1: 'btint4cmp(integer,integer)'}}
|
|
|
|
def test_map_operclass_default(self):
|
|
"Map a default operator class"
|
|
stmts = [CREATE_TYPE_STMT,
|
|
"CREATE OPERATOR CLASS oc1 DEFAULT FOR TYPE sd.myint "
|
|
"USING btree AS OPERATOR 1 <, OPERATOR 3 =, "
|
|
"FUNCTION 1 btmyintcmp(sd.myint,sd.myint)"]
|
|
dbmap = self.to_map(stmts)
|
|
assert dbmap['schema sd']['operator class oc1 using btree'] == \
|
|
{'type': 'myint', 'default': True,
|
|
'operators': {1: 'sd.<(sd.myint,sd.myint)',
|
|
3: 'sd.=(sd.myint,sd.myint)'},
|
|
'functions': {1: 'sd.btmyintcmp(sd.myint,sd.myint)'}}
|
|
|
|
def test_map_operclass_comment(self):
|
|
"Map an operator class comment"
|
|
dbmap = self.to_map([CREATE_STMT, COMMENT_STMT])
|
|
assert dbmap['schema sd']['operator class oc1 using btree'][
|
|
'description'] == 'Test operator class oc1'
|
|
|
|
|
|
class OperatorClassToSqlTestCase(InputMapToSqlTestCase):
|
|
"""Test SQL generation from input operators"""
|
|
|
|
@pytest.mark.xfail
|
|
def test_create_operclass1(self):
|
|
"Create an operator class"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'operator class oc1 using btree': {
|
|
'type': 'integer',
|
|
'operators': {1: '<(integer,integer)', 3: '=(integer,integer)'},
|
|
'functions': {1: 'btint4cmp(integer,integer)'}}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == CREATE_STMT_LONG
|
|
|
|
def test_create_operclass_default(self):
|
|
"Create a default operator class"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'operator class oc1 using btree': {
|
|
'default': True, 'type': 'myint',
|
|
'operators': {1: 'sd.<(sd.myint,sd.myint)',
|
|
3: 'sd.=(sd.myint,sd.myint)'},
|
|
'functions': {1: 'sd.btmyintcmp(sd.myint,sd.myint)'}}})
|
|
sql = self.to_sql(inmap, [CREATE_TYPE_STMT], superuser=True)
|
|
assert fix_indent(sql[0]) == "CREATE OPERATOR CLASS sd.oc1 DEFAULT " \
|
|
"FOR TYPE sd.myint USING btree AS OPERATOR 1 " \
|
|
"sd.<(sd.myint,sd.myint), OPERATOR 3 sd.=(sd.myint,sd.myint), "\
|
|
"FUNCTION 1 sd.btmyintcmp(sd.myint,sd.myint)"
|
|
|
|
@pytest.mark.xfail
|
|
def test_create_operclass_in_schema(self):
|
|
"Create a operator within a non-default schema"
|
|
inmap = self.std_map()
|
|
inmap.update({'schema s1': {'operator class oc1 using btree': {
|
|
'type': 'integer',
|
|
'operators': {1: '<(integer,integer)', 3: '=(integer,integer)'},
|
|
'functions': {1: 'btint4cmp(integer,integer)'}}}})
|
|
sql = self.to_sql(inmap, ["CREATE SCHEMA s1"])
|
|
assert fix_indent(sql[0]) == "CREATE OPERATOR CLASS s1.oc1 FOR " \
|
|
"TYPE integer USING btree AS OPERATOR 1 <(integer,integer), " \
|
|
"OPERATOR 3 =(integer,integer), " \
|
|
"FUNCTION 1 btint4cmp(integer,integer)"
|
|
|
|
def test_drop_operclass(self):
|
|
"Drop an existing operator"
|
|
sql = self.to_sql(self.std_map(), [CREATE_STMT], superuser=True)
|
|
assert sql == ["DROP OPERATOR CLASS sd.oc1 USING btree",
|
|
"DROP OPERATOR FAMILY sd.oc1 USING btree"]
|
|
|
|
@pytest.mark.xfail
|
|
def test_operclass_with_comment(self):
|
|
"Create an operator class with a comment"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'operator class oc1 using btree': {
|
|
'description': 'Test operator class oc1', 'type': 'integer',
|
|
'operators': {1: '<(integer,integer)', 3: '=(integer,integer)'},
|
|
'functions': {1: 'btint4cmp(integer,integer)'}}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == CREATE_STMT_LONG
|
|
assert sql[1] == COMMENT_STMT
|
|
|
|
def test_comment_on_operclass(self):
|
|
"Create a comment for an existing operator class"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'operator class oc1 using btree': {
|
|
'description': 'Test operator class oc1', 'type': 'integer',
|
|
'operators': {1: '<(integer,integer)', 3: '=(integer,integer)'},
|
|
'functions': {1: 'btint4cmp(integer,integer)'}},
|
|
'operator family oc1 using btree': {}})
|
|
sql = self.to_sql(inmap, [CREATE_STMT], superuser=True)
|
|
assert sql == [COMMENT_STMT]
|
|
|
|
def test_drop_operclass_comment(self):
|
|
"Drop the existing comment on an operator class"
|
|
stmts = [CREATE_STMT, COMMENT_STMT]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'operator class oc1 using btree': {
|
|
'type': 'integer',
|
|
'operators': {1: '<(integer,integer)', 3: '=(integer,integer)'},
|
|
'functions': {1: 'btint4cmp(integer,integer)'}},
|
|
'operator family oc1 using btree': {}})
|
|
sql = self.to_sql(inmap, stmts, superuser=True)
|
|
assert sql == ["COMMENT ON OPERATOR CLASS sd.oc1 USING btree IS NULL"]
|
|
|
|
def test_change_operclass_comment(self):
|
|
"Change existing comment on an operator class"
|
|
stmts = [CREATE_STMT, COMMENT_STMT]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'operator class oc1 using btree': {
|
|
'description': 'Changed operator class oc1', 'type': 'integer',
|
|
'operators': {1: '<(integer,integer)', 3: '=(integer,integer)'},
|
|
'functions': {1: 'btint4cmp(integer,integer)'}},
|
|
'operator family oc1 using btree': {}})
|
|
sql = self.to_sql(inmap, stmts, superuser=True)
|
|
assert sql == ["COMMENT ON OPERATOR CLASS sd.oc1 USING btree IS "
|
|
"'Changed operator class oc1'"]
|