1import os 2import time 3from collections import OrderedDict 4from datetime import ( 5 date as Date, 6 datetime as Datetime, 7 time as Time, 8 timedelta as Timedelta, 9 timezone as Timezone, 10) 11from decimal import Decimal 12from enum import Enum 13from ipaddress import IPv4Address, IPv4Network 14from json import dumps 15from uuid import UUID 16 17import pytest 18 19import pytz 20 21from pg8000.converters import ( 22 BIGINT, 23 BIGINT_ARRAY, 24 BOOLEAN, 25 CIDR_ARRAY, 26 DATE, 27 FLOAT_ARRAY, 28 INET, 29 INTEGER_ARRAY, 30 INTERVAL, 31 JSON, 32 JSONB, 33 JSONB_ARRAY, 34 JSON_ARRAY, 35 MONEY, 36 MONEY_ARRAY, 37 NUMERIC, 38 NUMERIC_ARRAY, 39 PGInterval, 40 POINT, 41 SMALLINT_ARRAY, 42 TIME, 43 TIMESTAMP, 44 TIMESTAMPTZ, 45 TIMESTAMPTZ_ARRAY, 46 TIMESTAMP_ARRAY, 47 UUID_ARRAY, 48 UUID_TYPE, 49 XID, 50 pg_interval_in, 51 pg_interval_out, 52 time_in, 53) 54 55 56def test_str_then_int(con): 57 v1 = "hello world" 58 retval = con.run("SELECT cast(:v1 as varchar) as f1", v1=v1) 59 assert retval[0][0] == v1 60 61 v2 = 1 62 retval = con.run("SELECT cast(:v2 as varchar) as f1", v2=v2) 63 assert retval[0][0] == str(v2) 64 65 66def test_insert_null(con): 67 v = None 68 con.run("CREATE TEMPORARY TABLE test_int (f INTEGER)") 69 con.run("INSERT INTO test_int VALUES (:v)", v=v) 70 retval = con.run("SELECT * FROM test_int") 71 assert retval[0][0] == v 72 73 74def test_int_roundtrip(con): 75 int2 = 21 76 int4 = 23 77 int8 = 20 78 79 MAP = { 80 int2: "int2", 81 int4: "int4", 82 int8: "int8", 83 } 84 85 test_values = [ 86 (0, int2), 87 (-32767, int2), 88 (-32768, int4), 89 (+32767, int2), 90 (+32768, int4), 91 (-2147483647, int4), 92 (-2147483648, int8), 93 (+2147483647, int4), 94 (+2147483648, int8), 95 (-9223372036854775807, int8), 96 (+9223372036854775807, int8), 97 ] 98 99 for value, typoid in test_values: 100 retval = con.run("SELECT cast(:v as " + MAP[typoid] + ")", v=value) 101 assert retval[0][0] == value 102 column_type_oid = con.columns[0]["type_oid"] 103 assert column_type_oid == typoid 104 105 106def test_bytearray_subclass_round_trip(con): 107 class BClass(bytearray): 108 pass 109 110 binary = b"\x00\x01\x02\x03\x02\x01\x00" 111 retval = con.run("SELECT cast(:v as bytea)", v=BClass(binary)) 112 assert retval[0][0] == binary 113 114 115def test_timestamp_roundtrip(is_java, con): 116 v = Datetime(2001, 2, 3, 4, 5, 6, 170000) 117 retval = con.run("SELECT cast(:v as timestamp)", v=v) 118 assert retval[0][0] == v 119 120 # Test that time zone doesn't affect it 121 # Jython 2.5.3 doesn't have a time.tzset() so skip 122 if not is_java: 123 orig_tz = os.environ.get("TZ") 124 os.environ["TZ"] = "America/Edmonton" 125 time.tzset() 126 127 retval = con.run("SELECT cast(:v as timestamp)", v=v) 128 assert retval[0][0] == v 129 130 if orig_tz is None: 131 del os.environ["TZ"] 132 else: 133 os.environ["TZ"] = orig_tz 134 time.tzset() 135 136 137def test_interval_roundtrip(con): 138 con.register_in_adapter(INTERVAL, pg_interval_in) 139 con.register_out_adapter(PGInterval, pg_interval_out) 140 v = PGInterval(microseconds=123456789, days=2, months=24) 141 retval = con.run("SELECT cast(:v as interval)", v=v) 142 assert retval[0][0] == v 143 144 145def test_enum_str_round_trip(con): 146 try: 147 con.run("create type lepton as enum ('electron', 'muon', 'tau')") 148 149 v = "muon" 150 retval = con.run("SELECT cast(:v as lepton) as f1", v=v) 151 assert retval[0][0] == v 152 con.run("CREATE TEMPORARY TABLE testenum (f1 lepton)") 153 con.run("INSERT INTO testenum VALUES (cast(:v as lepton))", v="electron") 154 finally: 155 con.run("drop table testenum") 156 con.run("drop type lepton") 157 158 159def test_enum_custom_round_trip(con): 160 class Lepton: 161 # Implements PEP 435 in the minimal fashion needed 162 __members__ = OrderedDict() 163 164 def __init__(self, name, value, alias=None): 165 self.name = name 166 self.value = value 167 self.__members__[name] = self 168 setattr(self.__class__, name, self) 169 if alias: 170 self.__members__[alias] = self 171 setattr(self.__class__, alias, self) 172 173 def lepton_out(lepton): 174 return lepton.value 175 176 try: 177 con.run("create type lepton as enum ('1', '2', '3')") 178 con.register_out_adapter(Lepton, lepton_out) 179 180 v = Lepton("muon", "2") 181 retval = con.run("SELECT CAST(:v AS lepton)", v=v) 182 assert retval[0][0] == v.value 183 finally: 184 con.run("drop type lepton") 185 186 187def test_enum_py_round_trip(con): 188 class Lepton(Enum): 189 electron = "1" 190 muon = "2" 191 tau = "3" 192 193 try: 194 con.run("create type lepton as enum ('1', '2', '3')") 195 196 v = Lepton.muon 197 retval = con.run("SELECT cast(:v as lepton) as f1", v=v) 198 assert retval[0][0] == v.value 199 200 con.run("CREATE TEMPORARY TABLE testenum (f1 lepton)") 201 con.run("INSERT INTO testenum VALUES (cast(:v as lepton))", v=Lepton.electron) 202 finally: 203 con.run("drop table testenum") 204 con.run("drop type lepton") 205 206 207def test_xml_roundtrip(con): 208 v = "<genome>gatccgagtac</genome>" 209 retval = con.run("select xmlparse(content :v) as f1", v=v) 210 assert retval[0][0] == v 211 212 213def test_int2vector_in(con): 214 retval = con.run("select cast('1 2' as int2vector) as f1") 215 assert retval[0][0] == [1, 2] 216 217 # Should complete without an exception 218 con.run("select indkey from pg_index") 219 220 221def test_timestamp_tz_out(con): 222 retval = con.run( 223 "SELECT '2001-02-03 04:05:06.17 America/Edmonton'" "::timestamp with time zone" 224 ) 225 dt = retval[0][0] 226 assert dt.tzinfo is not None, "no tzinfo returned" 227 assert dt.astimezone(Timezone.utc) == Datetime( 228 2001, 2, 3, 11, 5, 6, 170000, Timezone.utc 229 ), "retrieved value match failed" 230 231 232def test_timestamp_tz_roundtrip(is_java, con): 233 if not is_java: 234 mst = pytz.timezone("America/Edmonton") 235 v1 = mst.localize(Datetime(2001, 2, 3, 4, 5, 6, 170000)) 236 retval = con.run("SELECT cast(:v as timestamptz)", v=v1) 237 v2 = retval[0][0] 238 assert v2.tzinfo is not None 239 assert v1 == v2 240 241 242def test_timestamp_mismatch(is_java, con): 243 if not is_java: 244 mst = pytz.timezone("America/Edmonton") 245 con.run("SET SESSION TIME ZONE 'America/Edmonton'") 246 try: 247 con.run( 248 "CREATE TEMPORARY TABLE TestTz (f1 timestamp with time zone, " 249 "f2 timestamp without time zone)" 250 ) 251 con.run( 252 "INSERT INTO TestTz (f1, f2) VALUES (:v1, :v2)", 253 # insert timestamp into timestamptz field (v1) 254 v1=Datetime(2001, 2, 3, 4, 5, 6, 170000), 255 # insert timestamptz into timestamp field (v2) 256 v2=mst.localize(Datetime(2001, 2, 3, 4, 5, 6, 170000)), 257 ) 258 retval = con.run("SELECT f1, f2 FROM TestTz") 259 260 # when inserting a timestamp into a timestamptz field, 261 # postgresql assumes that it is in local time. So the value 262 # that comes out will be the server's local time interpretation 263 # of v1. We've set the server's TZ to MST, the time should 264 # be... 265 f1 = retval[0][0] 266 assert f1 == Datetime(2001, 2, 3, 11, 5, 6, 170000, Timezone.utc) 267 268 # inserting the timestamptz into a timestamp field, pg8000 converts the 269 # value into UTC, and then the PG server sends that time back 270 f2 = retval[0][1] 271 assert f2 == Datetime(2001, 2, 3, 11, 5, 6, 170000) 272 finally: 273 con.run("SET SESSION TIME ZONE DEFAULT") 274 275 276def test_name_out(con): 277 # select a field that is of "name" type: 278 con.run("SELECT usename FROM pg_user") 279 # It is sufficient that no errors were encountered. 280 281 282def test_oid_out(con): 283 con.run("SELECT oid FROM pg_type") 284 # It is sufficient that no errors were encountered. 285 286 287def test_boolean_in(con): 288 retval = con.run("SELECT cast('t' as bool)") 289 assert retval[0][0] 290 291 292def test_numeric_out(con): 293 for num in ("5000", "50.34"): 294 retval = con.run("SELECT " + num + "::numeric") 295 assert str(retval[0][0]) == num 296 297 298def test_int2_out(con): 299 retval = con.run("SELECT 5000::smallint") 300 assert retval[0][0] == 5000 301 302 303def test_int4_out(con): 304 retval = con.run("SELECT 5000::integer") 305 assert retval[0][0] == 5000 306 307 308def test_int8_out(con): 309 retval = con.run("SELECT 50000000000000::bigint") 310 assert retval[0][0] == 50000000000000 311 312 313def test_float4_out(con): 314 retval = con.run("SELECT 1.1::real") 315 assert retval[0][0] == 1.1 316 317 318def test_float8_out(con): 319 retval = con.run("SELECT 1.1::double precision") 320 assert retval[0][0] == 1.1000000000000001 321 322 323def test_varchar_out(con): 324 retval = con.run("SELECT 'hello'::varchar(20)") 325 assert retval[0][0] == "hello" 326 327 328def test_char_out(con): 329 retval = con.run("SELECT 'hello'::char(20)") 330 assert retval[0][0] == "hello " 331 332 333def test_text_out(con): 334 retval = con.run("SELECT 'hello'::text") 335 assert retval[0][0] == "hello" 336 337 338def test_pg_interval_in(con): 339 con.register_in_adapter(1186, pg_interval_in) 340 retval = con.run( 341 "SELECT CAST('1 month 16 days 12 hours 32 minutes 64 seconds' as INTERVAL)" 342 ) 343 expected_value = PGInterval( 344 microseconds=(12 * 60 * 60 * 1000 * 1000) 345 + (32 * 60 * 1000 * 1000) 346 + (64 * 1000 * 1000), 347 days=16, 348 months=1, 349 ) 350 assert retval[0][0] == expected_value 351 352 353def test_interval_in_30_seconds(con): 354 retval = con.run("select interval '30 seconds'") 355 assert retval[0][0] == Timedelta(seconds=30) 356 357 358def test_interval_in_12_days_30_seconds(con): 359 retval = con.run("select interval '12 days 30 seconds'") 360 assert retval[0][0] == Timedelta(days=12, seconds=30) 361 362 363def test_timestamp_out(con): 364 retval = con.run("SELECT '2001-02-03 04:05:06.17'::timestamp") 365 assert retval[0][0] == Datetime(2001, 2, 3, 4, 5, 6, 170000) 366 367 368def test_int4_array_out(con): 369 retval = con.run( 370 "SELECT '{1,2,3,4}'::INT[] AS f1, '{{1,2,3},{4,5,6}}'::INT[][] AS f2, " 371 "'{{{1,2},{3,4}},{{NULL,6},{7,8}}}'::INT[][][] AS f3" 372 ) 373 f1, f2, f3 = retval[0] 374 assert f1 == [1, 2, 3, 4] 375 assert f2 == [[1, 2, 3], [4, 5, 6]] 376 assert f3 == [[[1, 2], [3, 4]], [[None, 6], [7, 8]]] 377 378 379def test_int2_array_out(con): 380 res = con.run( 381 "SELECT '{1,2,3,4}'::INT2[] AS f1, " 382 "'{{1,2,3},{4,5,6}}'::INT2[][] AS f2, " 383 "'{{{1,2},{3,4}},{{NULL,6},{7,8}}}'::INT2[][][] AS f3" 384 ) 385 f1, f2, f3 = res[0] 386 assert f1 == [1, 2, 3, 4] 387 assert f2 == [[1, 2, 3], [4, 5, 6]] 388 assert f3 == [[[1, 2], [3, 4]], [[None, 6], [7, 8]]] 389 390 391def test_int8_array_out(con): 392 res = con.run( 393 "SELECT '{1,2,3,4}'::INT8[] AS f1, " 394 "'{{1,2,3},{4,5,6}}'::INT8[][] AS f2, " 395 "'{{{1,2},{3,4}},{{NULL,6},{7,8}}}'::INT8[][][] AS f3" 396 ) 397 f1, f2, f3 = res[0] 398 assert f1 == [1, 2, 3, 4] 399 assert f2 == [[1, 2, 3], [4, 5, 6]] 400 assert f3 == [[[1, 2], [3, 4]], [[None, 6], [7, 8]]] 401 402 403def test_bool_array_out(con): 404 res = con.run( 405 "SELECT '{TRUE,FALSE,FALSE,TRUE}'::BOOL[] AS f1, " 406 "'{{TRUE,FALSE,TRUE},{FALSE,TRUE,FALSE}}'::BOOL[][] AS f2, " 407 "'{{{TRUE,FALSE},{FALSE,TRUE}},{{NULL,TRUE},{FALSE,FALSE}}}'" 408 "::BOOL[][][] AS f3" 409 ) 410 f1, f2, f3 = res[0] 411 assert f1 == [True, False, False, True] 412 assert f2 == [[True, False, True], [False, True, False]] 413 assert f3 == [[[True, False], [False, True]], [[None, True], [False, False]]] 414 415 416def test_float4_array_out(con): 417 res = con.run( 418 "SELECT '{1,2,3,4}'::FLOAT4[] AS f1, " 419 "'{{1,2,3},{4,5,6}}'::FLOAT4[][] AS f2, " 420 "'{{{1,2},{3,4}},{{NULL,6},{7,8}}}'::FLOAT4[][][] AS f3" 421 ) 422 f1, f2, f3 = res[0] 423 assert f1 == [1, 2, 3, 4] 424 assert f2 == [[1, 2, 3], [4, 5, 6]] 425 assert f3 == [[[1, 2], [3, 4]], [[None, 6], [7, 8]]] 426 427 428def test_float8_array_out(con): 429 res = con.run( 430 "SELECT '{1,2,3,4}'::FLOAT8[] AS f1, " 431 "'{{1,2,3},{4,5,6}}'::FLOAT8[][] AS f2, " 432 "'{{{1,2},{3,4}},{{NULL,6},{7,8}}}'::FLOAT8[][][] AS f3" 433 ) 434 f1, f2, f3 = res[0] 435 assert f1 == [1, 2, 3, 4] 436 assert f2 == [[1, 2, 3], [4, 5, 6]] 437 assert f3 == [[[1, 2], [3, 4]], [[None, 6], [7, 8]]] 438 439 440CURRENCIES = { 441 "en_GB.UTF-8": "£", 442 "C.UTF-8": "$", 443 "C.UTF8": "$", 444} 445LANG = os.environ["LANG"] 446CURRENCY = CURRENCIES[LANG] 447 448 449@pytest.mark.parametrize( 450 "test_input,oid", 451 [ 452 [[Datetime(2001, 2, 3, 4, 5, 6)], TIMESTAMP_ARRAY], # timestamp[] 453 [ # timestamptz[] 454 [Datetime(2001, 2, 3, 4, 5, 6, 0, Timezone.utc)], 455 TIMESTAMPTZ_ARRAY, 456 ], 457 [ 458 {"name": "Apollo 11 Cave", "zebra": True, "age": 26.003}, 459 # json 460 JSON, 461 ], 462 [{"name": "Apollo 11 Cave", "zebra": True, "age": 26.003}, JSONB], # jsonb 463 [[IPv4Network("192.168.0.0/28")], CIDR_ARRAY], # cidr[] 464 [[1, 2, 3], SMALLINT_ARRAY], # int2[] 465 [[[1, 2], [3, 4]], SMALLINT_ARRAY], # int2[] multidimensional 466 [[1, None, 3], INTEGER_ARRAY], # int4[] with None 467 [[7000000000, 2, 3], BIGINT_ARRAY], # int8[] 468 [[1.1, 2.2, 3.3], FLOAT_ARRAY], # float8[] 469 [[Decimal("1.1"), None, Decimal("3.3")], NUMERIC_ARRAY], # numeric[] 470 [[f"{CURRENCY}1.10", None, f"{CURRENCY}3.30"], MONEY_ARRAY], # money[] 471 [[UUID("911460f2-1f43-fea2-3e2c-e01fd5b5069d")], UUID_ARRAY], # uuid[] 472 [ # json[] 473 [{"name": "Apollo 11 Cave", "zebra": True, "age": 26.003}], 474 JSON_ARRAY, 475 ], 476 [ # jsonb[] 477 [{"name": "Apollo 11 Cave", "zebra": True, "age": 26.003}], 478 JSONB_ARRAY, 479 ], 480 [Time(4, 5, 6), TIME], # time 481 [Date(2001, 2, 3), DATE], # date 482 [Datetime(2001, 2, 3, 4, 5, 6), TIMESTAMP], # timestamp 483 [Datetime(2001, 2, 3, 4, 5, 6, 0, Timezone.utc), TIMESTAMPTZ], # timestamptz 484 [True, BOOLEAN], # bool 485 [None, BOOLEAN], # null 486 [Decimal("1.1"), NUMERIC], # numeric 487 [f"{CURRENCY}1.10", MONEY], # money 488 [f"-{CURRENCY}1.10", MONEY], # money 489 [50000000000000, BIGINT], # int8 490 [UUID("911460f2-1f43-fea2-3e2c-e01fd5b5069d"), UUID_TYPE], # uuid 491 [IPv4Network("192.168.0.0/28"), INET], # inet 492 [IPv4Address("192.168.0.1"), INET], # inet 493 [86722, XID], # xid 494 ["infinity", TIMESTAMP], # timestamp 495 ["(2.3,1)", POINT], # point 496 [{"name": "Apollo 11 Cave", "zebra": True, "age": 26.003}, JSON], # json 497 [{"name": "Apollo 11 Cave", "zebra": True, "age": 26.003}, JSONB], # jsonb 498 ], 499) 500def test_roundtrip_oid(con, test_input, oid): 501 502 retval = con.run("SELECT :v", v=test_input, types={"v": oid}) 503 assert retval[0][0] == test_input 504 505 assert oid == con.columns[0]["type_oid"] 506 507 508@pytest.mark.parametrize( 509 "test_input,typ,required_version", 510 [ 511 [[True, False, None], "bool[]", None], 512 [[IPv4Address("192.168.0.1")], "inet[]", None], 513 [[Date(2021, 3, 1)], "date[]", None], 514 [[Datetime(2001, 2, 3, 4, 5, 6)], "timestamp[]", None], 515 [[Datetime(2001, 2, 3, 4, 5, 6, 0, Timezone.utc)], "timestamptz[]", None], 516 [[Time(4, 5, 6)], "time[]", None], 517 [[Timedelta(seconds=30)], "interval[]", None], 518 [[{"name": "Apollo 11 Cave", "zebra": True, "age": 26.003}], "jsonb[]", None], 519 [[b"\x00\x01\x02\x03\x02\x01\x00"], "bytea[]", None], 520 [[Decimal("1.1"), None, Decimal("3.3")], "numeric[]", None], 521 [[UUID("911460f2-1f43-fea2-3e2c-e01fd5b5069d")], "uuid[]", None], 522 [ 523 [ 524 "Hello!", 525 "World!", 526 "abcdefghijklmnopqrstuvwxyz", 527 "", 528 "A bunch of random characters:", 529 " ~!@#$%^&*()_+`1234567890-=[]\\{}|{;':\",./<>?\t", 530 None, 531 ], 532 "varchar[]", 533 None, 534 ], 535 [Timedelta(seconds=30), "interval", None], 536 [Time(4, 5, 6), "time", None], 537 [Date(2001, 2, 3), "date", None], 538 [Datetime(2001, 2, 3, 4, 5, 6), "timestamp", None], 539 [Datetime(2001, 2, 3, 4, 5, 6, 0, Timezone.utc), "timestamptz", None], 540 [True, "bool", 10], 541 [Decimal("1.1"), "numeric", None], 542 [1.756e-12, "float8", None], 543 [float("inf"), "float8", None], 544 ["hello world", "unknown", 10], 545 ["hello \u0173 world", "varchar", 10], 546 [50000000000000, "int8", None], 547 [b"\x00\x01\x02\x03\x02\x01\x00", "bytea", None], 548 [bytearray(b"\x00\x01\x02\x03\x02\x01\x00"), "bytea", None], 549 [UUID("911460f2-1f43-fea2-3e2c-e01fd5b5069d"), "uuid", None], 550 [IPv4Network("192.168.0.0/28"), "inet", None], 551 [IPv4Address("192.168.0.1"), "inet", None], 552 ], 553) 554def test_roundtrip_cast(con, pg_version, test_input, typ, required_version): 555 if required_version is not None and pg_version < required_version: 556 return 557 558 retval = con.run(f"SELECT CAST(:v AS {typ})", v=test_input) 559 assert retval[0][0] == test_input 560 561 562@pytest.mark.parametrize( 563 "test_input,expected", 564 [ 565 ("SELECT CAST('{a,b,c}' AS TEXT[])", ["a", "b", "c"]), 566 ("SELECT CAST('{a,b,c}' AS CHAR[])", ["a", "b", "c"]), 567 ("SELECT CAST('{a,b,c}' AS VARCHAR[])", ["a", "b", "c"]), 568 ("SELECT CAST('{a,b,c}' AS CSTRING[])", ["a", "b", "c"]), 569 ("SELECT CAST('{a,b,c}' AS NAME[])", ["a", "b", "c"]), 570 ("SELECT CAST('{}' AS text[])", []), 571 ('SELECT CAST(\'{NULL,"NULL",NULL,""}\' AS text[])', [None, "NULL", None, ""]), 572 ], 573) 574def test_array_in(con, test_input, expected): 575 result = con.run(test_input) 576 assert result[0][0] == expected 577 578 579def test_numeric_array_out(con): 580 res = con.run("SELECT '{1.1,2.2,3.3}'::numeric[] AS f1") 581 assert res[0][0] == [Decimal("1.1"), Decimal("2.2"), Decimal("3.3")] 582 583 584def test_empty_array(con): 585 v = [] 586 retval = con.run("SELECT cast(:v as varchar[])", v=v) 587 assert retval[0][0] == v 588 589 590def test_macaddr(con): 591 retval = con.run("SELECT macaddr '08002b:010203'") 592 assert retval[0][0] == "08:00:2b:01:02:03" 593 594 595def test_tsvector_roundtrip(con): 596 retval = con.run( 597 "SELECT cast(:v as tsvector)", v="a fat cat sat on a mat and ate a fat rat" 598 ) 599 assert retval[0][0] == "'a' 'and' 'ate' 'cat' 'fat' 'mat' 'on' 'rat' 'sat'" 600 601 602def test_hstore_roundtrip(con): 603 val = '"a"=>"1"' 604 retval = con.run("SELECT cast(:val as hstore)", val=val) 605 assert retval[0][0] == val 606 607 608def test_json_access_object(con): 609 val = {"name": "Apollo 11 Cave", "zebra": True, "age": 26.003} 610 retval = con.run("SELECT cast(:val as json) -> :name", val=dumps(val), name="name") 611 assert retval[0][0] == "Apollo 11 Cave" 612 613 614def test_jsonb_access_object(con): 615 val = {"name": "Apollo 11 Cave", "zebra": True, "age": 26.003} 616 retval = con.run("SELECT cast(:val as jsonb) -> :name", val=dumps(val), name="name") 617 assert retval[0][0] == "Apollo 11 Cave" 618 619 620def test_json_access_array(con): 621 val = [-1, -2, -3, -4, -5] 622 retval = con.run( 623 "SELECT cast(:v1 as json) -> cast(:v2 as int)", v1=dumps(val), v2=2 624 ) 625 assert retval[0][0] == -3 626 627 628def test_jsonb_access_array(con): 629 val = [-1, -2, -3, -4, -5] 630 retval = con.run( 631 "SELECT cast(:v1 as jsonb) -> cast(:v2 as int)", v1=dumps(val), v2=2 632 ) 633 assert retval[0][0] == -3 634 635 636def test_jsonb_access_path(con): 637 j = {"a": [1, 2, 3], "b": [4, 5, 6]} 638 639 path = ["a", "2"] 640 641 retval = con.run("SELECT cast(:v1 as jsonb) #>> :v2", v1=dumps(j), v2=path) 642 assert retval[0][0] == str(j[path[0]][int(path[1])]) 643 644 645def test_time_in(): 646 actual = time_in("12:57:18.000396") 647 assert actual == Time(12, 57, 18, 396) 648