--

-- Test various corner cases.

--

--- COERCION ---

-- This attempts to validate that _handler is properly handling relayed exceptions

CREATE OR REPLACE FUNCTION check_failed_srf_coercion() RETURNS setof int LANGUAGE python AS

$python$

def main():

return ((),(),())

$python$;

SELECT check_failed_srf_coercion();

SELECT * FROM check_failed_srf_coercion();

-- This will trigger a failure from int's typinput [due to str()]

CREATE OR REPLACE FUNCTION check_failed_srf_coercion2() RETURNS setof int LANGUAGE python AS

$python$

def main():

return "foobar"

$python$;

SELECT check_failed_srf_coercion2();

SELECT * FROM check_failed_srf_coercion2();

--- EXCEPTIONS ---

-- The 'selekt 2' exception is the one represented on exit.

-- However, the 'selekt 1' exception *must* be printed in the context.

CREATE OR REPLACE FUNCTION check_proper_pg_exc_printing() RETURNS setof int LANGUAGE python AS

$python$

def main():

try:

with xact():

prepare('selekt 1')

except:

prepare('selekt 2')

$python$;

SELECT check_proper_pg_exc_printing();

--- numeric boundaries ---

CREATE OR REPLACE FUNCTION check_numeric_edges() RETURNS text LANGUAGE python AS

$python$

import Postgres

int2 = Postgres.types.int2

int4 = Postgres.types.int4

int8 = Postgres.types.int8

int2_min = -(2**15)

int2_max = (2**15)-1

int4_min = -(2**31)

int4_max = (2**31)-1

int8_min = -(2**63)

int8_max = (2**63)-1

def main():

assert int(int2(int2_min)) == int2_min

assert int(int2(int2_max)) == int2_max

assert int(int4(int4_min)) == int4_min

assert int(int4(int4_max)) == int4_max

assert int(int8(int8_min)) == int8_min

assert int(int8(int8_max)) == int8_max

return 'success'

$python$;

SELECT check_numeric_edges();

CREATE OR REPLACE FUNCTION check_numeric_outside_edges() RETURNS text LANGUAGE python AS

$python$

import Postgres

int2 = Postgres.types.int2

int4 = Postgres.types.int4

int8 = Postgres.types.int8

int2_min = -(2**15)

int2_max = (2**15)-1

int4_min = -(2**31)

int4_max = (2**31)-1

int8_min = -(2**63)

int8_max = (2**63)-1

trials = [

(int2, int2_min - 1, int2_max + 1),

(int4, int4_min - 1, int4_max + 1),

(int8, int8_min - 1, int8_max + 1)

]

def main():

for x in trials:

typ, min, max = x

try:

with xact():

typ(min)

raise RuntimeError("%s failed to throw exception for (min) %d" %(typ.typname, min))

except (Postgres.Exception, OverflowError):

pass

try:

with xact():

typ(max)

raise RuntimeError("%s failed to throw exception for (max) %d" %(typ.typname, max))

except (Postgres.Exception, OverflowError):

pass

return 'success'

$python$;

SELECT check_numeric_outside_edges();

-- CURSORS --

CREATE OR REPLACE FUNCTION whence_overflow() RETURNS text LANGUAGE python AS

$python$

def main():

ps = prepare('select 1')

c = ps.declare()

try:

# might need to increase this number on some platforms? =)

c.seek(0, 2**128 + 1)

except OverflowError:

pass

return 'success'

$python$;

SELECT whence_overflow();

-- SRFs --

-- validate key errors

CREATE OR REPLACE FUNCTION cause_key_error_srf(OUT i int, OUT t text) RETURNS SETOF record LANGUAGE python AS

$python$

def main():

return (

{'f': 'NOOOOOO'},

)

$python$;

SELECT cause_key_error_srf();

SELECT * FROM cause_key_error_srf();

--- between transaction deallocation ---

CREATE OR REPLACE FUNCTION check_between_xact_deallocs() RETURNS text LANGUAGE python AS

$python$

# we want the state to be cleared when

# transaction scope is dumped.

from Postgres import WARNING, Stateful

@Stateful

def main():

p = prepare('select 1')

c = p.rows()

yield 'setup refs'

$python$;

-- can't validate deallocation without weakref support

-- I manually validated that dealloc (via elog) was getting called -jwp@2009

SELECT check_between_xact_deallocs();

SELECT check_between_xact_deallocs();

--- Direct Function Calls ---

-- This varchar_cast will return it's *exact* argument Datum.

-- This is validating that Postgres.Function is operating properly in these conditions.

CREATE OR REPLACE FUNCTION check_dont_free_params() RETURNS text LANGUAGE python AS

$python$

import Postgres

import gc

# yeah, hardcoded Oid is lame. Fix if you like.

varchar_cast = Postgres.Function(669)

def main():

param = Postgres.types.varchar("foo")

r = varchar_cast(param, -1, False)

assert r is not param

del r

del param

# just to be sure...

gc.collect()

return 'success'

$python$;

SELECT check_dont_free_params();

--- Array Type Invalidation ---

DROP TABLE IF EXISTS array_element_t;

CREATE TABLE array_element_t (i int, t text);

CREATE OR REPLACE FUNCTION check_new_array_type() RETURNS text LANGUAGE python AS

$python$

import Postgres

from Postgres.types import regtype

array_element_t = Postgres.Type(regtype('array_element_t'))

array = array_element_t.Array

def main():

assert Postgres.Type(array.oid) is array

sqlexec('ALTER TABLE array_element_t ADD COLUMN x float')

new_array = Postgres.Type(array.oid)

# IsCurrent should identify that the element type has changed

# and cause a new type to be created and used.

assert new_array is not array

assert new_array.Element is not array_element_t

return 'success'

$python$;

SELECT check_new_array_type();

--- Held Cursors ---

--

-- This case is testing how function refresh works

-- in the held cursor context. Functions will be refreshed when

-- the xid has changed. This *has* to happen as the transaction scope

-- has been reset, and the objects in fn_info whose references were held

-- by the transaction scope are no longer safe to use.

--

-- XXX: ATM, the cursor is being fully materialized, so as-is this

-- doesn't test anything. However, leave the test here for documentation

-- purposes and the ability to note when a change happens. If someone

-- knows how to force PG to not materialize the result set, this test

-- would likely benefit from that..

--

CREATE OR REPLACE FUNCTION replace_while_held(i int) RETURNS text VOLATILE LANGUAGE python AS

$python$

def main(i):

return 'foo'

$python$;

DECLARE held SCROLL CURSOR WITH HOLD FOR SELECT replace_while_held(i) FROM generate_series(1, 11) AS g(i) WHERE random() >= 0;

FETCH 5 IN held;

CREATE OR REPLACE FUNCTION replace_while_held(i int) RETURNS text VOLATILE LANGUAGE python AS

$python$

def main(i):

return 'bar'

$python$;

-- show that it's bar.

SELECT replace_while_held(1);

-- things are "fine", it's still foo. (because it was already materialized)

FETCH ALL IN held;

CLOSE held;