from pgcli.packages.sqlcompletion import (
suggest_type,
Special,
Database,
Schema,
Table,
Column,
View,
Keyword,
FromClauseItem,
Function,
Datatype,
Alias,
JoinCondition,
Join,
)
from pgcli.packages.parseutils.tables import TableReference
import pytest
def cols_etc(
table, schema=None, alias=None, is_function=False, parent=None, last_keyword=None
):
"""Returns the expected select-clause suggestions for a single-table
select."""
return set(
[
Column(
table_refs=(TableReference(schema, table, alias, is_function),),
qualifiable=True,
),
Function(schema=parent),
Keyword(last_keyword),
]
)
def test_select_suggests_cols_with_visible_table_scope():
suggestions = suggest_type("SELECT FROM tabl", "SELECT ")
assert set(suggestions) == cols_etc("tabl", last_keyword="SELECT")
def test_select_suggests_cols_with_qualified_table_scope():
suggestions = suggest_type("SELECT FROM sch.tabl", "SELECT ")
assert set(suggestions) == cols_etc("tabl", "sch", last_keyword="SELECT")
def test_cte_does_not_crash():
sql = "WITH CTE AS (SELECT F.* FROM Foo F WHERE F.Bar > 23) SELECT C.* FROM CTE C WHERE C.FooID BETWEEN 123 AND 234;"
for i in range(len(sql)):
suggestions = suggest_type(sql[: i + 1], sql[: i + 1])
@pytest.mark.parametrize("expression", ['SELECT * FROM "tabl" WHERE '])
def test_where_suggests_columns_functions_quoted_table(expression):
expected = cols_etc("tabl", alias='"tabl"', last_keyword="WHERE")
suggestions = suggest_type(expression, expression)
assert expected == set(suggestions)
@pytest.mark.parametrize(
"expression",
[
"INSERT INTO OtherTabl(ID, Name) SELECT * FROM tabl WHERE ",
"INSERT INTO OtherTabl SELECT * FROM tabl WHERE ",
"SELECT * FROM tabl WHERE ",
"SELECT * FROM tabl WHERE (",
"SELECT * FROM tabl WHERE foo = ",
"SELECT * FROM tabl WHERE bar OR ",
"SELECT * FROM tabl WHERE foo = 1 AND ",
"SELECT * FROM tabl WHERE (bar > 10 AND ",
"SELECT * FROM tabl WHERE (bar AND (baz OR (qux AND (",
"SELECT * FROM tabl WHERE 10 < ",
"SELECT * FROM tabl WHERE foo BETWEEN ",
"SELECT * FROM tabl WHERE foo BETWEEN foo AND ",
],
)
def test_where_suggests_columns_functions(expression):
suggestions = suggest_type(expression, expression)
assert set(suggestions) == cols_etc("tabl", last_keyword="WHERE")
@pytest.mark.parametrize(
"expression",
["SELECT * FROM tabl WHERE foo IN (", "SELECT * FROM tabl WHERE foo IN (bar, "],
)
def test_where_in_suggests_columns(expression):
suggestions = suggest_type(expression, expression)
assert set(suggestions) == cols_etc("tabl", last_keyword="WHERE")
@pytest.mark.parametrize("expression", ["SELECT 1 AS ", "SELECT 1 FROM tabl AS "])
def test_after_as(expression):
suggestions = suggest_type(expression, expression)
assert set(suggestions) == set()
def test_where_equals_any_suggests_columns_or_keywords():
text = "SELECT * FROM tabl WHERE foo = ANY("
suggestions = suggest_type(text, text)
assert set(suggestions) == cols_etc("tabl", last_keyword="WHERE")
def test_lparen_suggests_cols_and_funcs():
suggestion = suggest_type("SELECT MAX( FROM tbl", "SELECT MAX(")
assert set(suggestion) == set(
[
Column(table_refs=((None, "tbl", None, False),), qualifiable=True),
Function(schema=None),
Keyword("("),
]
)
def test_select_suggests_cols_and_funcs():
suggestions = suggest_type("SELECT ", "SELECT ")
assert set(suggestions) == set(
[
Column(table_refs=(), qualifiable=True),
Function(schema=None),
Keyword("SELECT"),
]
)
@pytest.mark.parametrize(
"expression", ["INSERT INTO ", "COPY ", "UPDATE ", "DESCRIBE "]
)
def test_suggests_tables_views_and_schemas(expression):
suggestions = suggest_type(expression, expression)
assert set(suggestions) == set([Table(schema=None), View(schema=None), Schema()])
@pytest.mark.parametrize("expression", ["SELECT * FROM "])
def test_suggest_tables_views_schemas_and_functions(expression):
suggestions = suggest_type(expression, expression)
assert set(suggestions) == set([FromClauseItem(schema=None), Schema()])
@pytest.mark.parametrize(
"expression",
[
"SELECT * FROM foo JOIN bar on bar.barid = foo.barid JOIN ",
"SELECT * FROM foo JOIN bar USING (barid) JOIN ",
],
)
def test_suggest_after_join_with_two_tables(expression):
suggestions = suggest_type(expression, expression)
tables = tuple([(None, "foo", None, False), (None, "bar", None, False)])
assert set(suggestions) == set(
[FromClauseItem(schema=None, table_refs=tables), Join(tables, None), Schema()]
)
@pytest.mark.parametrize(
"expression", ["SELECT * FROM foo JOIN ", "SELECT * FROM foo JOIN bar"]
)
def test_suggest_after_join_with_one_table(expression):
suggestions = suggest_type(expression, expression)
tables = ((None, "foo", None, False),)
assert set(suggestions) == set(
[
FromClauseItem(schema=None, table_refs=tables),
Join(((None, &qu