1
1
mirror of https://github.com/dbcli/pgcli.git synced 2024-11-09 16:34:32 +03:00

Fix PEP8 empty line errors

This commit is contained in:
Étienne BERSAC 2017-06-14 17:06:04 +02:00
parent f4d6adfc2c
commit e124c03575
No known key found for this signature in database
GPG Key ID: 95997557AD5A6DBF
26 changed files with 87 additions and 28 deletions

View File

@ -100,34 +100,41 @@ def refresher(name, refreshers=CompletionRefresher.refreshers):
return wrapped
return wrapper
@refresher('schemata')
def refresh_schemata(completer, executor):
completer.set_search_path(executor.search_path())
completer.extend_schemata(executor.schemata())
@refresher('tables')
def refresh_tables(completer, executor):
completer.extend_relations(executor.tables(), kind='tables')
completer.extend_columns(executor.table_columns(), kind='tables')
completer.extend_foreignkeys(executor.foreignkeys())
@refresher('views')
def refresh_views(completer, executor):
completer.extend_relations(executor.views(), kind='views')
completer.extend_columns(executor.view_columns(), kind='views')
@refresher('functions')
def refresh_functions(completer, executor):
completer.extend_functions(executor.functions())
@refresher('types')
def refresh_types(completer, executor):
completer.extend_datatypes(executor.datatypes())
@refresher('databases')
def refresh_databases(completer, executor):
completer.extend_database_names(executor.databases())
@refresher('casing')
def refresh_casing(completer, executor):
casing_file = completer.casing_file

View File

@ -3,6 +3,7 @@ import sys
PY2 = sys.version_info[0] == 2
PY3 = sys.version_info[0] == 3
def unicode2utf8(arg):
"""
Only in Python 2. Psycopg2 expects the args as bytes not unicode.
@ -13,6 +14,7 @@ def unicode2utf8(arg):
return arg.encode('utf-8')
return arg
def utf8tounicode(arg):
"""
Only in Python 2. Psycopg2 returns the error message as utf-8.

View File

@ -1,5 +1,6 @@
from prompt_toolkit.filters import Filter
class HasSelectedCompletion(Filter):
"""Enable when the current buffer has a selected completion."""

View File

@ -58,12 +58,5 @@ def pgcli_line_magic(line):
_logger.debug('Dangerous query detected -- ignoring')
return
ipython = get_ipython()
return ipython.run_cell_magic('sql', line, q.query)

View File

@ -1,9 +1,11 @@
from .tabulate import _text_type
from ..encodingutils import utf8tounicode
def pad(field, total, char=u" "):
return field + (char * (total - len(field)))
def expanded_table(rows, headers, missingval=""):
header_len = max([len(x) for x in headers])
max_row_len = 0

View File

@ -22,6 +22,7 @@
from UserDict import DictMixin
class OrderedDict(dict, DictMixin):
def __init__(self, *args, **kwds):

View File

@ -140,9 +140,3 @@ def _identifiers(tok):
yield t
elif isinstance(tok, Identifier):
yield tok

View File

@ -15,6 +15,7 @@ cleanup_regex = {
'all_punctuations': re.compile('([^\s]+)$'),
}
def last_word(text, include='alphanum_underscore'):
"""
Find the last word in a sentence.
@ -137,4 +138,3 @@ def parse_partial_identifier(word):
return parse_partial_identifier(word + '"')
else:
return None

View File

@ -1,2 +0,0 @@

View File

@ -14,6 +14,7 @@ def _compile_regex(keyword):
pattern = '\\b' + white_space_regex.sub(r'\\s+', keyword) + '\\b'
return re.compile(pattern, re.MULTILINE | re.IGNORECASE)
keywords = get_literals('keywords')
keyword_regexs = dict((kw, _compile_regex(kw)) for kw in keywords)
@ -48,6 +49,3 @@ class PrevalenceCounter(object):
def name_count(self, name):
return self.name_counts[name]

View File

@ -170,6 +170,7 @@ def _strip_named_query(txt):
txt = named_query_regex.sub('', txt)
return txt
function_body_pattern = re.compile('(\\$.*?\\$)([\s\S]*?)\\1', re.M)
@ -236,6 +237,7 @@ SPECIALS_SUGGESTION = {
'sf': Function,
}
def suggest_special(text):
text = text.lstrip()
cmd, _, arg = parse_special_command(text)

View File

@ -311,6 +311,7 @@ def _isint(string):
"""
return type(string) is _int_type or type(string) is _long_type
def _type(string, has_invisible=True):
"""The least generic type (type(None), int, float, str, unicode).

View File

@ -30,6 +30,8 @@ NamedQueries.instance = NamedQueries.from_config(
Match = namedtuple('Match', ['completion', 'priority'])
_SchemaObject = namedtuple('SchemaObject', ['name', 'schema', 'function'])
def SchemaObject(name, schema=None, function=False):
return _SchemaObject(name, schema, function)
@ -50,6 +52,7 @@ def generate_alias(tbl):
return ''.join([l for l in tbl if l.isupper()] or
[l for l, prev in zip(tbl, '_' + tbl) if prev == '_' and l != '_'])
class PGCompleter(Completer):
# keywords_tree: A dict mapping keywords to well known following keywords.
# e.g. 'CREATE': ['TABLE', 'USER', ...],
@ -827,6 +830,3 @@ class PGCompleter(Completer):
for meta in metas
if filter_func(meta)
]

View File

@ -2,6 +2,7 @@ from pygments.token import Token
from prompt_toolkit.enums import DEFAULT_BUFFER
from prompt_toolkit.key_binding.vi_state import InputMode
def _get_vi_mode(cli):
return {
InputMode.INSERT: 'I',
@ -10,6 +11,7 @@ def _get_vi_mode(cli):
InputMode.INSERT_MULTIPLE: 'M',
}[cli.vi_state.input_mode]
def create_toolbar_tokens_func(get_vi_mode_enabled, get_is_refreshing,
failed_transaction, valid_transaction):
"""

View File

@ -29,5 +29,3 @@ def executor(connection):
@pytest.fixture
def exception_formatter():
return lambda e: str(e)

View File

@ -21,6 +21,7 @@ def step_run_cli(context):
def step_wait_prompt(context):
wrappers.wait_prompt(context)
@when('we send "ctrl + d"')
def step_ctrl_d(context):
"""

View File

@ -117,5 +117,3 @@ def test_multiple_cte_extraction():
assert tuple(ctes) == (
('x', ('abc', 'def'), start1, stop1),
('y', ('ghi', 'jkl'), start2, stop2))

View File

@ -2,10 +2,12 @@ import pytest
from pgcli.packages.parseutils.tables import extract_tables
from pgcli.packages.parseutils.utils import find_prev_keyword, is_open_quote
def test_empty_string():
tables = extract_tables('')
assert tables == ()
def test_simple_select_single_table():
tables = extract_tables('select * from abc')
assert tables == ((None, 'abc', None, False),)
@ -33,62 +35,75 @@ def test_simple_select_single_table_double_quoted():
tables = extract_tables('select * from "Abc"')
assert tables == ((None, 'Abc', None, False),)
def test_simple_select_multiple_tables():
tables = extract_tables('select * from abc, def')
assert set(tables) == set([(None, 'abc', None, False),
(None, 'def', None, False)])
def test_simple_select_multiple_tables_double_quoted():
tables = extract_tables('select * from "Abc", "Def"')
assert set(tables) == set([(None, 'Abc', None, False),
(None, 'Def', None, False)])
def test_simple_select_single_table_deouble_quoted_aliased():
tables = extract_tables('select * from "Abc" a')
assert tables == ((None, 'Abc', 'a', False),)
def test_simple_select_multiple_tables_deouble_quoted_aliased():
tables = extract_tables('select * from "Abc" a, "Def" d')
assert set(tables) == set([(None, 'Abc', 'a', False),
(None, 'Def', 'd', False)])
def test_simple_select_multiple_tables_schema_qualified():
tables = extract_tables('select * from abc.def, ghi.jkl')
assert set(tables) == set([('abc', 'def', None, False),
('ghi', 'jkl', None, False)])
def test_simple_select_with_cols_single_table():
tables = extract_tables('select a,b from abc')
assert tables == ((None, 'abc', None, False),)
def test_simple_select_with_cols_single_table_schema_qualified():
tables = extract_tables('select a,b from abc.def')
assert tables == (('abc', 'def', None, False),)
def test_simple_select_with_cols_multiple_tables():
tables = extract_tables('select a,b from abc, def')
assert set(tables) == set([(None, 'abc', None, False),
(None, 'def', None, False)])
def test_simple_select_with_cols_multiple_qualified_tables():
tables = extract_tables('select a,b from abc.def, def.ghi')
assert set(tables) == set([('abc', 'def', None, False),
('def', 'ghi', None, False)])
def test_select_with_hanging_comma_single_table():
tables = extract_tables('select a, from abc')
assert tables == ((None, 'abc', None, False),)
def test_select_with_hanging_comma_multiple_tables():
tables = extract_tables('select a, from abc, def')
assert set(tables) == set([(None, 'abc', None, False),
(None, 'def', None, False)])
def test_select_with_hanging_period_multiple_tables():
tables = extract_tables('SELECT t1. FROM tabl1 t1, tabl2 t2')
assert set(tables) == set([(None, 'tabl1', 't1', False),
(None, 'tabl2', 't2', False)])
def test_simple_insert_single_table():
tables = extract_tables('insert into abc (id, name) values (1, "def")')
@ -98,19 +113,23 @@ def test_simple_insert_single_table():
assert tables == ((None, 'abc', 'abc', False),)
@pytest.mark.xfail
def test_simple_insert_single_table_schema_qualified():
tables = extract_tables('insert into abc.def (id, name) values (1, "def")')
assert tables == (('abc', 'def', None, False),)
def test_simple_update_table_no_schema():
tables = extract_tables('update abc set id = 1')
assert tables == ((None, 'abc', None, False),)
def test_simple_update_table_with_schema():
tables = extract_tables('update abc.def set id = 1')
assert tables == (('abc', 'def', None, False),)
@pytest.mark.parametrize('join_type', ['', 'INNER', 'LEFT', 'RIGHT OUTER'])
def test_join_table(join_type):
sql = 'SELECT * FROM abc a {0} JOIN def d ON a.id = d.num'.format(join_type)
@ -118,6 +137,7 @@ def test_join_table(join_type):
assert set(tables) == set([(None, 'abc', 'a', False),
(None, 'def', 'd', False)])
def test_join_table_schema_qualified():
tables = extract_tables('SELECT * FROM abc.def x JOIN ghi.jkl y ON x.id = y.num')
assert set(tables) == set([('abc', 'def', 'x', False),
@ -156,21 +176,25 @@ def test_subselect_tables():
tables = extract_tables(sql)
assert tables == ((None, 'abc', None, False),)
@pytest.mark.parametrize('text', ['SELECT * FROM foo.', 'SELECT 123 AS foo'])
def test_extract_no_tables(text):
tables = extract_tables(text)
assert tables == tuple()
@pytest.mark.parametrize('arg_list', ['', 'arg1', 'arg1, arg2, arg3'])
def test_simple_function_as_table(arg_list):
tables = extract_tables('SELECT * FROM foo({0})'.format(arg_list))
assert tables == ((None, 'foo', None, True),)
@pytest.mark.parametrize('arg_list', ['', 'arg1', 'arg1, arg2, arg3'])
def test_simple_schema_qualified_function_as_table(arg_list):
tables = extract_tables('SELECT * FROM foo.bar({0})'.format(arg_list))
assert tables == (('foo', 'bar', None, True),)
@pytest.mark.parametrize('arg_list', ['', 'arg1', 'arg1, arg2, arg3'])
def test_simple_aliased_function_as_table(arg_list):
tables = extract_tables('SELECT * FROM foo({0}) bar'.format(arg_list))
@ -182,6 +206,7 @@ def test_simple_table_and_function():
assert set(tables) == set([(None, 'foo', None, False),
(None, 'bar', None, True)])
def test_complex_table_and_function():
tables = extract_tables('''SELECT * FROM foo.bar baz
JOIN bar.qux(x, y, z) quux''')
@ -189,12 +214,12 @@ def test_complex_table_and_function():
('bar', 'qux', 'quux', True)])
def test_find_prev_keyword_using():
q = 'select * from tbl1 inner join tbl2 using (col1, '
kw, q2 = find_prev_keyword(q)
assert kw.value == '(' and q2 == 'select * from tbl1 inner join tbl2 using ('
@pytest.mark.parametrize('sql', [
'select * from foo where bar',
'select * from foo where bar = 1 and baz or ',

View File

@ -1,6 +1,7 @@
# coding=UTF-8
from pgcli.packages.expanded import expanded_table
def test_expanded_table_renders():
input = [("hello", 123),("world", 456)]
@ -13,6 +14,7 @@ age | 456
"""
assert expected == expanded_table(input, ["name", "age"])
def test_unicode_expanded_table():
input = [(u'ö', 123)]

View File

@ -73,6 +73,7 @@ def test_should_break_ties_using_lexical_order(completer, collection):
assert matches[1].priority > matches[0].priority
def test_matching_should_be_case_insensitive(completer):
"""Fuzzy matching should keep matches even if letter casing doesn't match.

View File

@ -55,6 +55,7 @@ def test_format_output():
expected = ['Title', '+---------+---------+\n| head1 | head2 |\n|---------+---------|\n| abc | def |\n+---------+---------+', 'test status']
assert results == expected
def test_format_output_auto_expand():
settings = OutputSettings(table_format='psql', dcmlfmt='d', floatfmt='g', max_width=100)
table_results = format_output('Title', [('abc', 'def')],

View File

@ -3,16 +3,19 @@ import pytest
from prompt_toolkit.completion import Completion
from prompt_toolkit.document import Document
@pytest.fixture
def completer():
import pgcli.pgcompleter as pgcompleter
return pgcompleter.PGCompleter(smart_completion=False)
@pytest.fixture
def complete_event():
from mock import Mock
return Mock()
def test_empty_string_completion(completer, complete_event):
text = ''
position = 0
@ -21,6 +24,7 @@ def test_empty_string_completion(completer, complete_event):
complete_event))
assert result == set(map(Completion, completer.all_completions))
def test_select_keyword_completion(completer, complete_event):
text = 'SEL'
position = len('SEL')
@ -29,6 +33,7 @@ def test_select_keyword_completion(completer, complete_event):
complete_event))
assert result == set([Completion(text='SELECT', start_position=-3)])
def test_function_name_completion(completer, complete_event):
text = 'SELECT MA'
position = len('SELECT MA')
@ -40,6 +45,7 @@ def test_function_name_completion(completer, complete_event):
Completion(text='MAX', start_position=-2),
Completion(text='MAXEXTENTS', start_position=-2)])
def test_column_name_completion(completer, complete_event):
text = 'SELECT FROM users'
position = len('SELECT ')
@ -48,6 +54,7 @@ def test_column_name_completion(completer, complete_event):
complete_event))
assert result == set(map(Completion, completer.all_completions))
def test_paths_completion(completer, complete_event):
text = '\i '
position = len(text)

View File

@ -20,6 +20,7 @@ def test_conn(executor):
+-----+
SELECT 1""")
@dbtest
def test_bools_are_treated_as_strings(executor):
run(executor, '''create table test(a boolean)''')
@ -32,6 +33,7 @@ def test_bools_are_treated_as_strings(executor):
+------+
SELECT 1""")
@dbtest
def test_schemata_table_views_and_columns_query(executor):
run(executor, "create table a(x text, y text)")
@ -63,6 +65,7 @@ def test_schemata_table_views_and_columns_query(executor):
assert set(executor.view_columns()) >= set([
('public', 'd', 'e', 'integer')])
@dbtest
def test_foreign_key_query(executor):
run(executor, "create schema schema1")
@ -73,6 +76,7 @@ def test_foreign_key_query(executor):
assert set(executor.foreignkeys()) >= set([
('schema1', 'parent', 'parentid', 'schema2', 'child', 'motherid')])
@dbtest
def test_functions_query(executor):
run(executor, '''create function func1() returns int
@ -108,17 +112,20 @@ def test_datatypes_query(executor):
types = list(executor.datatypes())
assert types == [('public', 'foo')]
@dbtest
def test_database_list(executor):
databases = executor.databases()
assert '_test_db' in databases
@dbtest
def test_invalid_syntax(executor, exception_formatter):
result = run(executor, 'invalid syntax!',
exception_formatter=exception_formatter)
assert 'syntax error at or near "invalid"' in result[0]
@dbtest
def test_invalid_column_name(executor, exception_formatter):
result = run(executor, 'select invalid command',
@ -148,6 +155,7 @@ def test_multiple_queries_same_line(executor):
assert "foo" in result[0]
assert "bar" in result[2]
@dbtest
def test_multiple_queries_with_special_command_same_line(executor, pgspecial):
result = run(executor, "select 'foo'; \d", pgspecial=pgspecial)
@ -156,6 +164,7 @@ def test_multiple_queries_with_special_command_same_line(executor, pgspecial):
# This is a lame check. :(
assert "Schema" in result[2]
@dbtest
def test_multiple_queries_same_line_syntaxerror(executor, exception_formatter):
result = run(executor, u"select 'fooé'; invalid syntax é",
@ -189,6 +198,7 @@ def test_bytea_field_support_in_output(executor):
def test_unicode_support_in_unknown_type(executor):
assert u'日本語' in run(executor, u"SELECT '日本語' AS japanese;", join=True)
@dbtest
def test_unicode_support_in_enum_type(executor):
run(executor, u"CREATE TYPE mood AS ENUM ('sad', 'ok', 'happy', '日本語')")
@ -196,6 +206,7 @@ def test_unicode_support_in_enum_type(executor):
run(executor, u"INSERT INTO person VALUES ('Moe', '日本語')")
assert u'日本語' in run(executor, u"SELECT * FROM person", join=True)
@requires_json
def test_json_renders_without_u_prefix(executor, expanded):
run(executor, u"create table jsontest(d json)")
@ -215,6 +226,7 @@ def test_jsonb_renders_without_u_prefix(executor, expanded):
assert u'{"name": "Éowyn"}' in result
@dbtest
def test_date_time_types(executor):
run(executor, "SET TIME ZONE UTC")
@ -231,6 +243,7 @@ def test_date_time_types(executor):
assert run(executor, "SELECT (CAST('-123456789 days 12:23:56' AS interval))", join=True).split("\n")[3] \
== "| -123456789 days, 12:23:56 |"
@dbtest
@pytest.mark.parametrize('value', ['10000000', '10000000.0', '10000000000000'])
def test_large_numbers_render_directly(executor, value):

View File

@ -547,6 +547,7 @@ def test_schema_object_order(completer):
table(t, pos=-1) for t in ('users', 'custom."Users"', 'custom.users')
]
@parametrize('completer', completers(casing=False, filtr=False, aliasing=False))
def test_all_schema_objects(completer):
text = ('SELECT * FROM ')

View File

@ -75,6 +75,7 @@ def test_builtin_function_name_completion(completer):
keyword('MAXEXTENTS', -2), keyword('MATERIALIZED VIEW', -2)
])
@parametrize('completer', completers())
def test_builtin_function_matches_only_at_start(completer):
text = 'SELECT IN'
@ -205,7 +206,6 @@ def test_suggested_multiple_column_names_with_alias(completer):
assert result == set(testdata.columns('users'))
@parametrize('completer', completers(casing=True))
def test_suggested_cased_column_names_with_alias(completer):
result = result_set(
@ -325,6 +325,7 @@ def test_suggested_joins_fuzzy(completer, text):
expected = join('users ON users.id = u.userid', -len(last_word))
assert expected in result
join_texts = [
'SELECT * FROM Users JOIN ',
'''INSERT INTO "Users"
@ -853,6 +854,7 @@ def test_keyword_casing_upper(keyword_casing, expected, texts):
completions = get_result(completer, text)
assert expected in [cpl.text for cpl in completions]
@parametrize('completer', completers())
def test_keyword_after_alter(completer):
text = 'ALTER TABLE users ALTER '

View File

@ -68,6 +68,7 @@ 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 ',
@ -212,6 +213,7 @@ def test_truncate_suggests_qualified_tables():
assert set(suggestions) == set([
Table(schema='sch')])
@pytest.mark.parametrize('text', [
'SELECT DISTINCT ',
'INSERT INTO foo SELECT DISTINCT '
@ -277,6 +279,7 @@ def test_distinct_and_order_by_suggestions_with_alias_given(text, text_before):
Function(schema='x'),
])
def test_col_comma_suggests_cols():
suggestions = suggest_type('SELECT a, b, FROM tbl', 'SELECT a, b,')
assert set(suggestions) == set([
@ -303,6 +306,7 @@ def test_into_suggests_tables_and_schemas():
Schema(),
])
@pytest.mark.parametrize('text', [
'INSERT INTO abc (',
'INSERT INTO abc () SELECT * FROM hij;',
@ -537,6 +541,7 @@ def test_on_suggests_aliases_and_join_conditions(sql):
assert set(suggestions) == set((JoinCondition(table_refs=tables, parent=None),
Alias(aliases=('a', 'b',)),))
@pytest.mark.parametrize('sql', [
'select abc.x, bcd.y from abc join bcd on abc.id = bcd.id AND ',
'select abc.x, bcd.y from abc join bcd on ',
@ -643,6 +648,7 @@ def test_3_statements_2nd_current():
'select * from a; select ')
assert set(suggestions) == cols_etc('b', last_keyword='SELECT')
@pytest.mark.parametrize('text', [
'''
CREATE OR REPLACE FUNCTION func() RETURNS setof int AS $$
@ -695,6 +701,7 @@ def test_statements_in_function_body(text):
Keyword('SELECT'),
])
functions = [
'''
CREATE OR REPLACE FUNCTION func() RETURNS setof int AS $$
@ -713,6 +720,7 @@ SELECT 1 FROM foo;
'''
]
@pytest.mark.parametrize('text', functions)
def test_statements_with_cursor_after_function_body(text):
suggestions = suggest_type(text, text[:text.find('; ') + 1])
@ -724,6 +732,7 @@ def test_statements_with_cursor_before_function_body(text):
suggestions = suggest_type(text, '')
assert set(suggestions) == set([Keyword(), Special()])
def test_create_db_with_template():
suggestions = suggest_type('create database foo with template ',
'create database foo with template ')