From 5ceb50e44906acb4de8610b8901030d27b75a322 Mon Sep 17 00:00:00 2001
From: Kaitmazian Maksim <m.kaitmazian@picodata.io>
Date: Fri, 27 Oct 2023 18:39:15 +0300
Subject: [PATCH] fix: failed lint-for-tests
---
pgproto/test/auth_test.py | 30 +++++---
pgproto/test/simple_query_test.py | 124 ++++++++++++++++++------------
2 files changed, 91 insertions(+), 63 deletions(-)
diff --git a/pgproto/test/auth_test.py b/pgproto/test/auth_test.py
index a967110e19..7a7972c683 100644
--- a/pgproto/test/auth_test.py
+++ b/pgproto/test/auth_test.py
@@ -1,17 +1,17 @@
import pytest
-import pg8000.dbapi as pg
+import pg8000.dbapi as pg # type: ignore
from conftest import Postgres
def test_auth(postgres: Postgres):
- host = '127.0.0.1'
+ host = "127.0.0.1"
port = 5432
postgres.start(host, port)
i1 = postgres.instance
- user = 'user'
- password = 'fANPIOUWEh79p12hdunqwADI'
+ user = "user"
+ password = "fANPIOUWEh79p12hdunqwADI"
i1.eval("box.cfg{auth_type='md5'}")
i1.call("pico.create_user", user, password, dict(timeout=3))
@@ -20,18 +20,24 @@ def test_auth(postgres: Postgres):
conn.close()
# test authentication with a wrong password
- with pytest.raises(pg.DatabaseError, match=f"authentication failed for user '{user}'"):
- pg.Connection(user, password='wrong password', host=host, port=port)
+ with pytest.raises(
+ 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
- with pytest.raises(pg.DatabaseError, match="authentication failed for user 'unknown-user'"):
- pg.Connection("unknown-user", password='aaa', host=host, port=port)
+ with pytest.raises(
+ 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_password = '231321fnijphui217h08'
+ sha_user = "chap-sha-enjoyer"
+ sha_password = "231321fnijphui217h08"
i1.eval("box.cfg{auth_type='chap-sha1'}")
i1.call("pico.create_user", sha_user, sha_password, dict(timeout=3))
# test authentication with an unsupported method
- with pytest.raises(pg.DatabaseError, match=f"authentication failed for user '{sha_user}'"):
- pg.Connection(sha_user, password='aaa', host=host, port=port)
+ with pytest.raises(
+ pg.DatabaseError, match=f"authentication failed for user '{sha_user}'"
+ ):
+ pg.Connection(sha_user, password="aaa", host=host, port=port)
diff --git a/pgproto/test/simple_query_test.py b/pgproto/test/simple_query_test.py
index fbe43862f7..4a4d781692 100644
--- a/pgproto/test/simple_query_test.py
+++ b/pgproto/test/simple_query_test.py
@@ -1,58 +1,65 @@
import pytest
-import pg8000.dbapi as pg
+import pg8000.dbapi as pg # type: ignore
import os
from conftest import Postgres
def test_simple_query_flow_errors(postgres: Postgres):
- host = '127.0.0.1'
+ host = "127.0.0.1"
port = 5432
postgres.start(host, port)
i1 = postgres.instance
- user = 'admin'
- password = 'fANPIOUWEh79p12hdunqwADI'
+ user = "admin"
+ password = "fANPIOUWEh79p12hdunqwADI"
i1.eval("box.cfg{auth_type='md5'}")
i1.eval(f"box.schema.user.passwd('{user}', '{password}')")
with pytest.raises(pg.InterfaceError, match="Server refuses SSL"):
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.autocommit = True
cur = conn.cursor()
- with pytest.raises(pg.DatabaseError, match="expected CreateUser, DropUser, CreateRole,"):
- cur.execute("""
+ with pytest.raises(
+ pg.DatabaseError, match="expected CreateUser, DropUser, CreateRole,"
+ ):
+ cur.execute(
+ """
CREATE TEMPORARY TABLE book (id SERIAL, title TEXT);
- """)
+ """
+ )
with pytest.raises(pg.DatabaseError, match="space BOOK not found"):
- cur.execute("""
+ cur.execute(
+ """
INSERT INTO book VALUES (1, 2);
- """)
+ """
+ )
def test_simple_flow_session(postgres: Postgres):
- host = '127.0.0.1'
+ host = "127.0.0.1"
port = 5432
postgres.start(host, port)
i1 = postgres.instance
- user = 'admin'
- password = 'password'
+ user = "admin"
+ password = "password"
i1.eval("box.cfg{auth_type='md5'}")
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.autocommit = True
cur = conn.cursor()
- cur.execute("""
+ cur.execute(
+ """
create table "tall" (
"id" integer not null,
"str" string,
@@ -62,54 +69,63 @@ def test_simple_flow_session(postgres: Postgres):
)
using memtx distributed by ("id")
option (timeout = 3);
- """)
+ """
+ )
- cur.execute("""
+ cur.execute(
+ """
INSERT INTO "tall" VALUES
(1, 'one', true, CAST(0.1 AS DOUBLE)),
(2, 'to', false, CAST(0.2 AS DOUBLE)),
(4, 'for', true, CAST(0.4 AS DOUBLE));
- """)
+ """
+ )
- cur.execute("""
+ cur.execute(
+ """
SELECT * FROM "tall";
- """)
+ """
+ )
tuples = cur.fetchall()
- assert [1, 'one', True, 0.1] in tuples
- assert [2, 'to', False, 0.2] in tuples
- assert [4, 'for', True, 0.4] in tuples
+ assert [1, "one", True, 0.1] in tuples
+ assert [2, "to", False, 0.2] in tuples
+ assert [4, "for", True, 0.4] in tuples
- cur.execute("""
+ cur.execute(
+ """
DROP TABLE "tall";
- """)
+ """
+ )
def test_explain(postgres: Postgres):
- host = '127.0.0.1'
+ host = "127.0.0.1"
port = 5432
postgres.start(host, port)
i1 = postgres.instance
- user = 'admin'
- password = 'password'
+ user = "admin"
+ password = "password"
i1.eval("box.cfg{auth_type='md5'}")
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.autocommit = True
cur = conn.cursor()
- cur.execute("""
+ cur.execute(
+ """
create table "explain" (
"id" integer not null,
primary key ("id")
)
using memtx distributed by ("id")
option (timeout = 3);
- """)
+ """
+ )
query = """
insert into "explain" values (0);
@@ -118,18 +134,18 @@ def test_explain(postgres: Postgres):
plan = cur.fetchall()
assert 'insert "explain" on conflict: fail' in plan[0]
assert ' motion [policy: local segment([ref("COLUMN_1")])]' in plan[1]
- assert ' values' in plan[2]
- assert ' value row (data=ROW(0::unsigned))' in plan[3]
- assert 'execution options:' in plan[4]
+ assert " values" in plan[2]
+ assert " value row (data=ROW(0::unsigned))" in plan[3]
+ assert "execution options:" in plan[4]
cur.execute(query)
cur.execute("explain " + query)
plan = cur.fetchall()
assert 'insert "explain" on conflict: fail' in plan[0]
assert ' motion [policy: local segment([ref("COLUMN_1")])]' in plan[1]
- assert ' values' in plan[2]
- assert ' value row (data=ROW(0::unsigned))' in plan[3]
- assert 'execution options:' in plan[4]
+ assert " values" in plan[2]
+ assert " value row (data=ROW(0::unsigned))" in plan[3]
+ assert "execution options:" in plan[4]
query = """
select * from "explain";
@@ -138,14 +154,14 @@ def test_explain(postgres: Postgres):
plan = cur.fetchall()
assert 'projection ("explain"."id"::integer -> "id")' in plan[0]
assert ' scan "explain"' in plan[1]
- assert 'execution options:' in plan[2]
+ assert "execution options:" in plan[2]
cur.execute(query)
cur.execute("explain " + query)
plan = cur.fetchall()
assert 'projection ("explain"."id"::integer -> "id")' in plan[0]
assert ' scan "explain"' in plan[1]
- assert 'execution options:' in plan[2]
+ assert "execution options:" in plan[2]
cur.execute('drop table "explain";')
@@ -153,23 +169,24 @@ def test_explain(postgres: Postgres):
# Aggregates return value type is decimal, which is currently not supported,
# so an error is expected.
def test_aggregate_error(postgres: Postgres):
- host = '127.0.0.1'
+ host = "127.0.0.1"
port = 5432
postgres.start(host, port)
i1 = postgres.instance
- user = 'admin'
- password = 'password'
+ user = "admin"
+ password = "password"
i1.eval("box.cfg{auth_type='md5'}")
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.autocommit = True
cur = conn.cursor()
- cur.execute("""
+ cur.execute(
+ """
create table "tall" (
"id" integer not null,
"str" string,
@@ -179,14 +196,19 @@ def test_aggregate_error(postgres: Postgres):
)
using memtx distributed by ("id")
option (timeout = 3);
- """)
+ """
+ )
- with pytest.raises(pg.DatabaseError, match="unknown column type \'decimal\'"):
- cur.execute("""
+ with pytest.raises(pg.DatabaseError, match="unknown column type 'decimal'"):
+ cur.execute(
+ """
SELECT SUM("id") FROM "tall";
- """)
+ """
+ )
- with pytest.raises(pg.DatabaseError, match="unknown column type \'decimal\'"):
- cur.execute("""
+ with pytest.raises(pg.DatabaseError, match="unknown column type 'decimal'"):
+ cur.execute(
+ """
SELECT AVG("id") FROM "tall";
- """)
+ """
+ )
--
GitLab