Skip to content
Snippets Groups Projects
Commit 5ceb50e4 authored by Maksim Kaitmazian's avatar Maksim Kaitmazian Committed by Maksim Kaitmazian
Browse files

fix: failed lint-for-tests

parent da629805
No related branches found
No related tags found
1 merge request!920pgproto module
import pytest import pytest
import pg8000.dbapi as pg import pg8000.dbapi as pg # type: ignore
from conftest import Postgres from conftest import Postgres
def test_auth(postgres: Postgres): def test_auth(postgres: Postgres):
host = '127.0.0.1' host = "127.0.0.1"
port = 5432 port = 5432
postgres.start(host, port) postgres.start(host, port)
i1 = postgres.instance i1 = postgres.instance
user = 'user' user = "user"
password = 'fANPIOUWEh79p12hdunqwADI' password = "fANPIOUWEh79p12hdunqwADI"
i1.eval("box.cfg{auth_type='md5'}") i1.eval("box.cfg{auth_type='md5'}")
i1.call("pico.create_user", user, password, dict(timeout=3)) i1.call("pico.create_user", user, password, dict(timeout=3))
...@@ -20,18 +20,24 @@ def test_auth(postgres: Postgres): ...@@ -20,18 +20,24 @@ def test_auth(postgres: Postgres):
conn.close() conn.close()
# test authentication with a wrong password # test authentication with a wrong password
with pytest.raises(pg.DatabaseError, match=f"authentication failed for user '{user}'"): with pytest.raises(
pg.Connection(user, password='wrong password', host=host, port=port) pg.DatabaseError, match=f"authentication failed for user '{user}'"
):
pg.Connection(user, password="wrong password", host=host, port=port)
# test authentication with an unknown user # test authentication with an unknown user
with pytest.raises(pg.DatabaseError, match="authentication failed for user 'unknown-user'"): with pytest.raises(
pg.Connection("unknown-user", password='aaa', host=host, port=port) pg.DatabaseError, match="authentication failed for user 'unknown-user'"
):
pg.Connection("unknown-user", password="aaa", host=host, port=port)
sha_user = 'chap-sha-enjoyer' sha_user = "chap-sha-enjoyer"
sha_password = '231321fnijphui217h08' sha_password = "231321fnijphui217h08"
i1.eval("box.cfg{auth_type='chap-sha1'}") i1.eval("box.cfg{auth_type='chap-sha1'}")
i1.call("pico.create_user", sha_user, sha_password, dict(timeout=3)) i1.call("pico.create_user", sha_user, sha_password, dict(timeout=3))
# test authentication with an unsupported method # test authentication with an unsupported method
with pytest.raises(pg.DatabaseError, match=f"authentication failed for user '{sha_user}'"): with pytest.raises(
pg.Connection(sha_user, password='aaa', host=host, port=port) pg.DatabaseError, match=f"authentication failed for user '{sha_user}'"
):
pg.Connection(sha_user, password="aaa", host=host, port=port)
import pytest import pytest
import pg8000.dbapi as pg import pg8000.dbapi as pg # type: ignore
import os import os
from conftest import Postgres from conftest import Postgres
def test_simple_query_flow_errors(postgres: Postgres): def test_simple_query_flow_errors(postgres: Postgres):
host = '127.0.0.1' host = "127.0.0.1"
port = 5432 port = 5432
postgres.start(host, port) postgres.start(host, port)
i1 = postgres.instance i1 = postgres.instance
user = 'admin' user = "admin"
password = 'fANPIOUWEh79p12hdunqwADI' password = "fANPIOUWEh79p12hdunqwADI"
i1.eval("box.cfg{auth_type='md5'}") i1.eval("box.cfg{auth_type='md5'}")
i1.eval(f"box.schema.user.passwd('{user}', '{password}')") i1.eval(f"box.schema.user.passwd('{user}', '{password}')")
with pytest.raises(pg.InterfaceError, match="Server refuses SSL"): with pytest.raises(pg.InterfaceError, match="Server refuses SSL"):
pg.Connection(user, password=password, host=host, port=port, ssl_context=True) pg.Connection(user, password=password, host=host, port=port, ssl_context=True)
os.environ['PGSSLMODE'] = 'disable' os.environ["PGSSLMODE"] = "disable"
conn = pg.Connection(user, password=password, host=host, port=port) conn = pg.Connection(user, password=password, host=host, port=port)
conn.autocommit = True conn.autocommit = True
cur = conn.cursor() cur = conn.cursor()
with pytest.raises(pg.DatabaseError, match="expected CreateUser, DropUser, CreateRole,"): with pytest.raises(
cur.execute(""" pg.DatabaseError, match="expected CreateUser, DropUser, CreateRole,"
):
cur.execute(
"""
CREATE TEMPORARY TABLE book (id SERIAL, title TEXT); CREATE TEMPORARY TABLE book (id SERIAL, title TEXT);
""") """
)
with pytest.raises(pg.DatabaseError, match="space BOOK not found"): with pytest.raises(pg.DatabaseError, match="space BOOK not found"):
cur.execute(""" cur.execute(
"""
INSERT INTO book VALUES (1, 2); INSERT INTO book VALUES (1, 2);
""") """
)
def test_simple_flow_session(postgres: Postgres): def test_simple_flow_session(postgres: Postgres):
host = '127.0.0.1' host = "127.0.0.1"
port = 5432 port = 5432
postgres.start(host, port) postgres.start(host, port)
i1 = postgres.instance i1 = postgres.instance
user = 'admin' user = "admin"
password = 'password' password = "password"
i1.eval("box.cfg{auth_type='md5'}") i1.eval("box.cfg{auth_type='md5'}")
i1.eval(f"box.schema.user.passwd('{user}', '{password}')") i1.eval(f"box.schema.user.passwd('{user}', '{password}')")
os.environ['PGSSLMODE'] = 'disable' os.environ["PGSSLMODE"] = "disable"
conn = pg.Connection(user, password=password, host=host, port=port) conn = pg.Connection(user, password=password, host=host, port=port)
conn.autocommit = True conn.autocommit = True
cur = conn.cursor() cur = conn.cursor()
cur.execute(""" cur.execute(
"""
create table "tall" ( create table "tall" (
"id" integer not null, "id" integer not null,
"str" string, "str" string,
...@@ -62,54 +69,63 @@ def test_simple_flow_session(postgres: Postgres): ...@@ -62,54 +69,63 @@ def test_simple_flow_session(postgres: Postgres):
) )
using memtx distributed by ("id") using memtx distributed by ("id")
option (timeout = 3); option (timeout = 3);
""") """
)
cur.execute(""" cur.execute(
"""
INSERT INTO "tall" VALUES INSERT INTO "tall" VALUES
(1, 'one', true, CAST(0.1 AS DOUBLE)), (1, 'one', true, CAST(0.1 AS DOUBLE)),
(2, 'to', false, CAST(0.2 AS DOUBLE)), (2, 'to', false, CAST(0.2 AS DOUBLE)),
(4, 'for', true, CAST(0.4 AS DOUBLE)); (4, 'for', true, CAST(0.4 AS DOUBLE));
""") """
)
cur.execute(""" cur.execute(
"""
SELECT * FROM "tall"; SELECT * FROM "tall";
""") """
)
tuples = cur.fetchall() tuples = cur.fetchall()
assert [1, 'one', True, 0.1] in tuples assert [1, "one", True, 0.1] in tuples
assert [2, 'to', False, 0.2] in tuples assert [2, "to", False, 0.2] in tuples
assert [4, 'for', True, 0.4] in tuples assert [4, "for", True, 0.4] in tuples
cur.execute(""" cur.execute(
"""
DROP TABLE "tall"; DROP TABLE "tall";
""") """
)
def test_explain(postgres: Postgres): def test_explain(postgres: Postgres):
host = '127.0.0.1' host = "127.0.0.1"
port = 5432 port = 5432
postgres.start(host, port) postgres.start(host, port)
i1 = postgres.instance i1 = postgres.instance
user = 'admin' user = "admin"
password = 'password' password = "password"
i1.eval("box.cfg{auth_type='md5'}") i1.eval("box.cfg{auth_type='md5'}")
i1.eval(f"box.schema.user.passwd('{user}', '{password}')") i1.eval(f"box.schema.user.passwd('{user}', '{password}')")
os.environ['PGSSLMODE'] = 'disable' os.environ["PGSSLMODE"] = "disable"
conn = pg.Connection(user, password=password, host=host, port=port) conn = pg.Connection(user, password=password, host=host, port=port)
conn.autocommit = True conn.autocommit = True
cur = conn.cursor() cur = conn.cursor()
cur.execute(""" cur.execute(
"""
create table "explain" ( create table "explain" (
"id" integer not null, "id" integer not null,
primary key ("id") primary key ("id")
) )
using memtx distributed by ("id") using memtx distributed by ("id")
option (timeout = 3); option (timeout = 3);
""") """
)
query = """ query = """
insert into "explain" values (0); insert into "explain" values (0);
...@@ -118,18 +134,18 @@ def test_explain(postgres: Postgres): ...@@ -118,18 +134,18 @@ def test_explain(postgres: Postgres):
plan = cur.fetchall() plan = cur.fetchall()
assert 'insert "explain" on conflict: fail' in plan[0] assert 'insert "explain" on conflict: fail' in plan[0]
assert ' motion [policy: local segment([ref("COLUMN_1")])]' in plan[1] assert ' motion [policy: local segment([ref("COLUMN_1")])]' in plan[1]
assert ' values' in plan[2] assert " values" in plan[2]
assert ' value row (data=ROW(0::unsigned))' in plan[3] assert " value row (data=ROW(0::unsigned))" in plan[3]
assert 'execution options:' in plan[4] assert "execution options:" in plan[4]
cur.execute(query) cur.execute(query)
cur.execute("explain " + query) cur.execute("explain " + query)
plan = cur.fetchall() plan = cur.fetchall()
assert 'insert "explain" on conflict: fail' in plan[0] assert 'insert "explain" on conflict: fail' in plan[0]
assert ' motion [policy: local segment([ref("COLUMN_1")])]' in plan[1] assert ' motion [policy: local segment([ref("COLUMN_1")])]' in plan[1]
assert ' values' in plan[2] assert " values" in plan[2]
assert ' value row (data=ROW(0::unsigned))' in plan[3] assert " value row (data=ROW(0::unsigned))" in plan[3]
assert 'execution options:' in plan[4] assert "execution options:" in plan[4]
query = """ query = """
select * from "explain"; select * from "explain";
...@@ -138,14 +154,14 @@ def test_explain(postgres: Postgres): ...@@ -138,14 +154,14 @@ def test_explain(postgres: Postgres):
plan = cur.fetchall() plan = cur.fetchall()
assert 'projection ("explain"."id"::integer -> "id")' in plan[0] assert 'projection ("explain"."id"::integer -> "id")' in plan[0]
assert ' scan "explain"' in plan[1] assert ' scan "explain"' in plan[1]
assert 'execution options:' in plan[2] assert "execution options:" in plan[2]
cur.execute(query) cur.execute(query)
cur.execute("explain " + query) cur.execute("explain " + query)
plan = cur.fetchall() plan = cur.fetchall()
assert 'projection ("explain"."id"::integer -> "id")' in plan[0] assert 'projection ("explain"."id"::integer -> "id")' in plan[0]
assert ' scan "explain"' in plan[1] assert ' scan "explain"' in plan[1]
assert 'execution options:' in plan[2] assert "execution options:" in plan[2]
cur.execute('drop table "explain";') cur.execute('drop table "explain";')
...@@ -153,23 +169,24 @@ def test_explain(postgres: Postgres): ...@@ -153,23 +169,24 @@ def test_explain(postgres: Postgres):
# Aggregates return value type is decimal, which is currently not supported, # Aggregates return value type is decimal, which is currently not supported,
# so an error is expected. # so an error is expected.
def test_aggregate_error(postgres: Postgres): def test_aggregate_error(postgres: Postgres):
host = '127.0.0.1' host = "127.0.0.1"
port = 5432 port = 5432
postgres.start(host, port) postgres.start(host, port)
i1 = postgres.instance i1 = postgres.instance
user = 'admin' user = "admin"
password = 'password' password = "password"
i1.eval("box.cfg{auth_type='md5'}") i1.eval("box.cfg{auth_type='md5'}")
i1.eval(f"box.schema.user.passwd('{user}', '{password}')") i1.eval(f"box.schema.user.passwd('{user}', '{password}')")
os.environ['PGSSLMODE'] = 'disable' os.environ["PGSSLMODE"] = "disable"
conn = pg.Connection(user, password=password, host=host, port=port) conn = pg.Connection(user, password=password, host=host, port=port)
conn.autocommit = True conn.autocommit = True
cur = conn.cursor() cur = conn.cursor()
cur.execute(""" cur.execute(
"""
create table "tall" ( create table "tall" (
"id" integer not null, "id" integer not null,
"str" string, "str" string,
...@@ -179,14 +196,19 @@ def test_aggregate_error(postgres: Postgres): ...@@ -179,14 +196,19 @@ def test_aggregate_error(postgres: Postgres):
) )
using memtx distributed by ("id") using memtx distributed by ("id")
option (timeout = 3); option (timeout = 3);
""") """
)
with pytest.raises(pg.DatabaseError, match="unknown column type \'decimal\'"): with pytest.raises(pg.DatabaseError, match="unknown column type 'decimal'"):
cur.execute(""" cur.execute(
"""
SELECT SUM("id") FROM "tall"; SELECT SUM("id") FROM "tall";
""") """
)
with pytest.raises(pg.DatabaseError, match="unknown column type \'decimal\'"): with pytest.raises(pg.DatabaseError, match="unknown column type 'decimal'"):
cur.execute(""" cur.execute(
"""
SELECT AVG("id") FROM "tall"; SELECT AVG("id") FROM "tall";
""") """
)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment