Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_u_suggests_databases():
suggestions = suggest_type("\\u ", "\\u ")
assert sorted_dicts(suggestions) == sorted_dicts([{"type": "database"}])
def test_describe_table():
suggestions = suggest_type("\\dt", "\\dt ")
assert sorted_dicts(suggestions) == sorted_dicts(
[
{"type": "table", "schema": []},
{"type": "view", "schema": []},
{"type": "schema"},
]
def test_list_or_show_create_tables():
suggestions = suggest_type("\\dt+", "\\dt+ ")
assert sorted_dicts(suggestions) == sorted_dicts(
[
{"type": "table", "schema": []},
{"type": "view", "schema": []},
{"type": "schema"},
]
def test_import_first_argument():
test_cases = [
# text, expecting_arg_idx
[".import ", 1],
[".import ./da", 1],
[".import ./data.csv ", 2],
[".import ./data.csv t", 2],
]
for text, expecting_arg_idx in test_cases:
suggestions = suggest_type(text, text)
if expecting_arg_idx == 1:
assert suggestions == [{"type": "file_name"}]
else:
assert suggestions == [{"type": "table", "schema": []}]
def get_completions(self, document, complete_event):
word_before_cursor = document.get_word_before_cursor(WORD=True)
completions = []
suggestions = suggest_type(document.text, document.text_before_cursor)
for suggestion in suggestions:
_logger.debug("Suggestion type: %r", suggestion["type"])
if suggestion["type"] == "column":
tables = suggestion["tables"]
_logger.debug("Completion column scope: %r", tables)
scoped_cols = self.populate_scoped_cols(tables)
if suggestion.get("drop_unique"):
# drop_unique is used for 'tb11 JOIN tbl2 USING (...'
# which should suggest only columns that appear in more than
# one table
scoped_cols = [
col
for (col, count) in Counter(scoped_cols).items()