Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@dbtest
def test_slash_d_verbose(executor):
results = executor('\d+')
title = None
rows = [('public', 'Inh1', 'table', POSTGRES_USER, '8192 bytes', None),
('public', 'inh2', 'table', POSTGRES_USER, '8192 bytes', None),
('public', 'mvw1', 'materialized view',
POSTGRES_USER, '8192 bytes', None),
('public', 'tbl1', 'table', POSTGRES_USER, '8192 bytes', None),
('public', 'tbl2', 'table', POSTGRES_USER, '8192 bytes', None),
('public', 'tbl2_id2_seq', 'sequence',
POSTGRES_USER, '8192 bytes', None),
('public', 'tbl3', 'table', POSTGRES_USER, '0 bytes', None),
('public', 'vw1', 'view', POSTGRES_USER, '0 bytes', None)]
headers = objects_listing_headers
status = 'SELECT 8'
expected = [title, rows, headers, status]
@dbtest
def test_slash_d_table_verbose_2(executor):
title = None
headers = ['Column', 'Type', 'Modifiers',
'Storage', 'Stats target', 'Description']
results = executor('\d+ tbl2')
rows = [['id2', 'integer', " not null default nextval('tbl2_id2_seq'::regclass)",
'plain', None, None],
['txt2', 'text', '', 'extended', None, None],
]
status = ('Child tables: inh2\n'
'Has OIDs: no\n')
expected = [title, rows, headers, status]
assert results == expected
results = executor('\d+ inh2')
@dbtest
def test_slash_h(executor):
"""List all commands."""
results = executor('\h')
expected = [None, help_rows, [], None]
assert results == expected
@dbtest
def test_slash_sf_parens(executor):
results = executor('\sf func1()')
title = None
rows = [('CREATE OR REPLACE FUNCTION public.func1()\n'
' RETURNS integer\n'
' LANGUAGE sql\n'
'AS $function$select 1$function$\n',),
]
headers = ['source']
status = None
expected = [title, rows, headers, status]
assert results == expected
@dbtest
def test_slash_copy_from_csv(executor, connection, tmpdir):
filepath = tmpdir.join('tbl1.csv')
executor("\copy (SELECT 22, 'elephant') TO '{0}' WITH csv"
.format(filepath))
executor("\copy tbl1 FROM '{0}' WITH csv".format(filepath))
cur = connection.cursor()
cur.execute("SELECT * FROM tbl1 WHERE id1 = 22")
row = cur.fetchone()
assert row[1] == 'elephant'
@dbtest
def test_slash_dx_verbose(executor):
"""List all extensions in verbose mode."""
results = executor('\dx+')
title = '\nObjects in extension "plpgsql"'
row = [('function plpgsql_call_handler()',),
('function plpgsql_inline_handler(internal)',),
('function plpgsql_validator(oid)',),
('language plpgsql',)]
headers = ['Object Description']
status = 'SELECT 4'
expected = [title, row, headers, status]
assert results == expected
@dbtest
def test_slash_db_name(executor):
"""List tablespace by name."""
title, rows, header, status = executor('\db pg_default')
assert title is None
assert header == ['Name', 'Owner', 'Location']
assert 'pg_default' in rows[0]
assert status == 'SELECT 1'
@dbtest
def test_slash_copy_to_tsv(executor, tmpdir):
filepath = tmpdir.join('pycons.tsv')
executor(u"\copy (SELECT 'Montréal', 'Portland', 'Cleveland') TO '{0}' "
.format(filepath))
infile = filepath.open(encoding='utf-8')
contents = infile.read()
assert len(contents.splitlines()) == 1
assert u'Montréal' in contents
@dbtest
def test_slash_copy_to_csv(executor, tmpdir):
filepath = tmpdir.join('pycons.tsv')
executor(u"\copy (SELECT 'Montréal', 'Portland', 'Cleveland') TO '{0}' WITH csv"
.format(filepath))
infile = filepath.open(encoding='utf-8')
contents = infile.read()
assert len(contents.splitlines()) == 1
assert u'Montréal' in contents
assert u',' in contents
@dbtest
def test_slash_d_table_verbose_1(executor):
title = None
headers = ['Column', 'Type', 'Modifiers',
'Storage', 'Stats target', 'Description']
results = executor('\d+ tbl1')
rows = [['id1', 'integer', ' not null', 'plain', None, None],
['txt1', 'text', ' not null', 'extended', None, None],
]
status = ('Indexes:\n "id_text" PRIMARY KEY, btree (id1, txt1)\n'
'Child tables: "Inh1",\n'
' inh2\n'
'Has OIDs: no\n')
expected = [title, rows, headers, status]
assert results == expected