File: test_specials.py

package info (click to toggle)
python-pgspecial 1.11.10%2Bdfsg1-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 352 kB
  • sloc: python: 3,143; makefile: 3
file content (627 lines) | stat: -rw-r--r-- 23,063 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
  #!/usr/bin/python
  # -*- coding: utf-8 -*-

from dbutils import dbtest, POSTGRES_USER, foreign_db_environ
import itertools

objects_listing_headers = ['Schema', 'Name', 'Type', 'Owner', 'Size', 'Description']


@dbtest
def test_slash_d(executor):
    results = executor('\d')
    title = None
    rows = [('public', 'Inh1', 'table', POSTGRES_USER),
            ('public', 'inh2', 'table', POSTGRES_USER),
            ('public', 'mvw1', 'materialized view', POSTGRES_USER),
            ('public', 'tbl1', 'table', POSTGRES_USER),
            ('public', 'tbl2', 'table', POSTGRES_USER),
            ('public', 'tbl2_id2_seq', 'sequence', POSTGRES_USER),
            ('public', 'tbl3', 'table', POSTGRES_USER),
            ('public', 'vw1', 'view', POSTGRES_USER)]
    headers = objects_listing_headers[:-2]
    status = 'SELECT 8'
    expected = [title, rows, headers, status]

    assert results == expected


@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]

    assert results == expected


@dbtest
def test_slash_d_table_1(executor):
    results = executor('\d tbl1')
    title = None
    rows = [['id1', 'integer', ' not null'],
            ['txt1', 'text', ' not null'],
            ]
    headers = ['Column', 'Type', 'Modifiers']
    status = ('Indexes:\n    "id_text" PRIMARY KEY, btree (id1, txt1)\n'
              'Number of child tables: 2 (Use \\d+ to list them.)\n')
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_d_table_2(executor):
    results = executor('\d tbl2')
    title = None
    rows = [['id2', 'integer', " not null default nextval('tbl2_id2_seq'::regclass)"],
            ['txt2', 'text', ''],
            ]
    headers = ['Column', 'Type', 'Modifiers']
    status = ('Number of child tables: 1 (Use \\d+ to list them.)\n')
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_d_test_generated_default(executor):
    results = executor('\d schema3.test_generated_default')
    headers = ['Column', 'Type', 'Modifiers']
    status = 'Indexes:\n    "test_generated_default_pkey" PRIMARY KEY, btree (id)\n'
    rows = [['id', 'integer', ' not null generated by default as identity'], [
        'some_stuff', 'text', '']]
    assert rows == results[1]
    assert headers == results[2]
    assert status == results[3]


@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

    results = executor('\d+ "Inh1"')
    rows = [['id1', 'integer', ' not null', 'plain', None, None],
            ['txt1', 'text', ' not null', 'extended', None, None],
            ['value1', 'integer', '', 'plain', None, None],
            ]
    status = ('Inherits: tbl1\n'
              'Has OIDs: no\n')
    expected = [title, rows, headers, status]
    assert results == expected


@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')
    rows = [['id1', 'integer', ' not null', 'plain', None, None],
            ['txt1', 'text', ' not null', 'extended', None, None],
            ['id2', 'integer', " not null default nextval('tbl2_id2_seq'::regclass)",
             'plain', None, None],
            ['txt2', 'text', '', 'extended', None, None],
            ['value2', 'integer', '', 'plain', None, None],
            ]
    status = ('Inherits: tbl1,\n'
              '          tbl2\n'
              'Has OIDs: no\n')
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_d_table_with_exclusion(executor):
    results = executor('\d tbl3')
    title = None
    rows = [['c3', 'circle', '']]
    headers = ['Column', 'Type', 'Modifiers']
    status = 'Indexes:\n    "tbl3_c3_excl" EXCLUDE USING gist (c3 WITH &&)\n'
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_d_table_2_in_schema(executor):
    results = executor('\d schema2.tbl2')
    title = None
    rows = [['id2', 'integer', " not null default nextval('schema2.tbl2_id2_seq'::regclass)"],
            ['txt2', 'text', ''],
            ]
    headers = ['Column', 'Type', 'Modifiers']
    status = ''
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_dn(executor):
    """List all schemas."""
    results = executor('\dn')
    title = None
    rows = [('public', POSTGRES_USER),
            ('schema1', POSTGRES_USER),
            ('schema2', POSTGRES_USER),
            ('schema3', POSTGRES_USER),
            ]
    headers = ['Name', 'Owner']
    status = 'SELECT %s' % len(rows)
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_dt(executor):
    """List all tables in public schema."""
    results = executor('\dt')
    title = None
    rows = [('public', 'Inh1', 'table', POSTGRES_USER),
            ('public', 'inh2', 'table', POSTGRES_USER),
            ('public', 'tbl1', 'table', POSTGRES_USER),
            ('public', 'tbl2', 'table', POSTGRES_USER),
            ('public', 'tbl3', 'table', POSTGRES_USER)]
    headers = objects_listing_headers[:-2]
    status = 'SELECT %s' % len(rows)
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_dt_verbose(executor):
    """List all tables in public schema in verbose mode."""
    results = executor('\dt+')
    title = None
    rows = [('public', 'Inh1', 'table', POSTGRES_USER, '8192 bytes', None),
            ('public', 'inh2', 'table', POSTGRES_USER, '8192 bytes', None),
            ('public', 'tbl1', 'table', POSTGRES_USER, '8192 bytes', None),
            ('public', 'tbl2', 'table', POSTGRES_USER, '8192 bytes', None),
            ('public', 'tbl3', 'table', POSTGRES_USER, '0 bytes', None)]
    headers = objects_listing_headers
    status = 'SELECT %s' % len(rows)
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_dv(executor):
    """List all views in public schema."""
    results = executor('\dv')
    title = None
    row = [('public', 'vw1', 'view', POSTGRES_USER)]
    headers = objects_listing_headers[:-2]
    status = 'SELECT 1'
    expected = [title, row, headers, status]
    assert results == expected


@dbtest
def test_slash_dv_verbose(executor):
    """List all views in s1 schema in verbose mode."""
    results = executor('\dv+ schema1.*')
    title = None
    row = [('schema1', 's1_vw1', 'view', POSTGRES_USER, '0 bytes', None)]
    headers = objects_listing_headers
    status = 'SELECT 1'
    expected = [title, row, headers, status]
    assert results == expected


@dbtest
def test_slash_dm(executor):
    """List all materialized views in schema1."""
    results = executor('\dm schema1.*')
    title = None
    row = [('schema1', 's1_mvw1', 'materialized view', POSTGRES_USER)]
    headers = objects_listing_headers[:-2]
    status = 'SELECT 1'
    expected = [title, row, headers, status]
    assert results == expected


@dbtest
def test_slash_dm_verbose(executor):
    """List all materialized views in public schema in verbose mode."""
    results = executor('\dm+')
    title = None
    row = [('public', 'mvw1', 'materialized view', POSTGRES_USER, '8192 bytes', None)]
    headers = objects_listing_headers
    status = 'SELECT 1'
    expected = [title, row, headers, status]
    assert results == expected


@dbtest
def test_slash_ds(executor):
    """List all sequences in public schema."""
    results = executor('\ds')
    title = None
    row = [('public', 'tbl2_id2_seq', 'sequence', POSTGRES_USER)]
    headers = objects_listing_headers[:-2]
    status = 'SELECT 1'
    expected = [title, row, headers, status]
    assert results == expected


@dbtest
def test_slash_ds_verbose(executor):
    """List all sequences in public schema in verbose mode."""
    results = executor('\ds+')
    title = None
    row = [('public', 'tbl2_id2_seq', 'sequence', POSTGRES_USER, '8192 bytes', None)]
    headers = objects_listing_headers
    status = 'SELECT 1'
    expected = [title, row, headers, status]
    assert results == expected


@dbtest
def test_slash_di(executor):
    """List all indexes in public schema."""
    results = executor('\di')
    title = None
    row = [('public', 'id_text', 'index', POSTGRES_USER),
           ('public', 'tbl3_c3_excl', 'index', POSTGRES_USER)]
    headers = objects_listing_headers[:-2]
    status = 'SELECT %s' % len(row)
    expected = [title, row, headers, status]
    assert results == expected


@dbtest
def test_slash_di_verbose(executor):
    """List all indexes in public schema in verbose mode."""
    results = executor('\di+')
    title = None
    row = [('public', 'id_text', 'index', POSTGRES_USER, '8192 bytes', None),
           ('public', 'tbl3_c3_excl', 'index', POSTGRES_USER, '8192 bytes', None)]
    headers = objects_listing_headers
    status = 'SELECT 2'
    expected = [title, row, headers, status]
    assert results == expected


@dbtest
def test_slash_dx(executor):
    """List all extensions."""
    results = executor('\dx')
    title = None
    row = [('plpgsql', '1.0', 'pg_catalog', 'PL/pgSQL procedural language')]
    headers = ['Name', 'Version', 'Schema', 'Description']
    status = 'SELECT 1'
    expected = [title, row, headers, status]
    assert results == expected


@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 %s' % len(row)
    expected = [title, row, headers, status]
    assert results == expected


@dbtest
def test_slash_dT(executor):
    """List all datatypes."""
    results = executor('\dT')
    title = None
    rows = [('public', 'foo', None),
            ('public', 'gender_t', None)]
    headers = ['Schema', 'Name', 'Description']
    status = 'SELECT %s' % len(rows)
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_dD(executor):
    title = None
    headers = ['Schema', 'Name', 'Type', 'Modifier', 'Check']
    results = executor('\dD')
    rows = [('public', 'gender_t', 'character(1)', '',
             "CHECK (VALUE = ANY (ARRAY['F'::bpchar, 'M'::bpchar]))")]
    status = 'SELECT 1'
    expected = [title, rows, headers, status]
    assert results == expected

    results = executor('\dD schema1.*')
    rows = [('schema1', 'bigint_t', 'bigint', '', ''),
            ('schema1', 'smallint_t', 'smallint', '', '')]
    status = 'SELECT %s' % len(rows)
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_dD_verbose(executor):
    title = None
    headers = ['Schema', 'Name', 'Type', 'Modifier', 'Check',
               'Access privileges', 'Description']
    results = executor('\dD+')
    rows = [('public', 'gender_t', 'character(1)', '',
             "CHECK (VALUE = ANY (ARRAY['F'::bpchar, 'M'::bpchar]))",
             None, None)]
    status = 'SELECT 1'
    expected = [title, rows, headers, status]
    assert results == expected

    results = executor('\dD+ schema1.bigint_t')
    rows = [('schema1', 'bigint_t', 'bigint', '', '', None,
             'a really large integer')]
    status = 'SELECT 1'
    expected = [title, rows, headers, status]
    assert results == expected


@dbtest
def test_slash_dF(executor):
    title, rows, header, status = executor('\dF')
    assert title is None
    assert header == ['Schema', 'Name', 'Description']
    assert ('pg_catalog', 'spanish',
            'configuration for spanish language') in rows

    results = executor('\dD *ian')
    assert title is None
    assert header == ['Schema', 'Name', 'Description']
    assert ('pg_catalog', 'russian',
            'configuration for russian language') in rows

    results = executor('\dD ge*')
    assert title is None
    assert header == ['Schema', 'Name', 'Description']
    assert ('pg_catalog', 'german', 'configuration for german language') in rows


@dbtest
def test_slash_dF_verbose(executor):
    results = executor('\dF+')[1]
    assert ('asciihword', 'simple') in results

    results = executor('\dF+ *panish')[1]
    assert ('asciihword', 'spanish_stem') in results

    results = executor('\dF+ swed*')[1]
    assert ('asciihword', 'swedish_stem') in results

    results = executor('\dF+ jap')
    assert results == [None, None, None,
                       'Did not find any results for pattern "jap".']

@dbtest
def test_slash_db(executor):
    """List all tablespaces."""
    title, rows, header, status = executor('\db')
    assert title is None
    assert header == ['Name', 'Owner', 'Location']
    assert 'pg_default' in rows[0]


@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_df(executor):
    results = executor('\df')
    title = None
    rows = [('public', 'func1', 'integer', '', 'normal')]
    headers = ['Schema', 'Name', 'Result data type', 'Argument data types',
            'Type']
    status = 'SELECT 1'
    expected = [title, rows, headers, status]
    assert results == expected

help_rows = [['ABORT', 'ALTER AGGREGATE', 'ALTER COLLATION', 'ALTER CONVERSION', 'ALTER DATABASE', 'ALTER DEFAULT PRIVILEGES'], ['ALTER DOMAIN', 'ALTER EVENT TRIGGER', 'ALTER EXTENSION', 'ALTER FOREIGN DATA WRAPPER', 'ALTER FOREIGN TABLE', 'ALTER FUNCTION'], ['ALTER GROUP', 'ALTER INDEX', 'ALTER LANGUAGE', 'ALTER LARGE OBJECT', 'ALTER MATERIALIZED VIEW', 'ALTER OPCLASS'], ['ALTER OPERATOR', 'ALTER OPFAMILY', 'ALTER POLICY', 'ALTER ROLE', 'ALTER RULE', 'ALTER SCHEMA'], ['ALTER SEQUENCE', 'ALTER SERVER', 'ALTER SYSTEM', 'ALTER TABLE', 'ALTER TABLESPACE', 'ALTER TRIGGER'], ['ALTER TSCONFIG', 'ALTER TSDICTIONARY', 'ALTER TSPARSER', 'ALTER TSTEMPLATE', 'ALTER TYPE', 'ALTER USER'], ['ALTER USER MAPPING', 'ALTER VIEW', 'ANALYZE', 'BEGIN', 'CHECKPOINT', 'CLOSE'], ['CLUSTER', 'COMMENT', 'COMMIT', 'COMMIT PREPARED', 'COPY', 'CREATE AGGREGATE'], ['CREATE CAST', 'CREATE COLLATION', 'CREATE CONVERSION', 'CREATE DATABASE', 'CREATE DOMAIN', 'CREATE EVENT TRIGGER'], ['CREATE EXTENSION', 'CREATE FOREIGN DATA WRAPPER', 'CREATE FOREIGN TABLE', 'CREATE FUNCTION', 'CREATE GROUP', 'CREATE INDEX'], ['CREATE LANGUAGE', 'CREATE MATERIALIZED VIEW', 'CREATE OPCLASS', 'CREATE OPERATOR', 'CREATE OPFAMILY', 'CREATE POLICY'], ['CREATE ROLE', 'CREATE RULE', 'CREATE SCHEMA', 'CREATE SEQUENCE', 'CREATE SERVER', 'CREATE TABLE'], ['CREATE TABLE AS', 'CREATE TABLESPACE', 'CREATE TRANSFORM', 'CREATE TRIGGER', 'CREATE TSCONFIG', 'CREATE TSDICTIONARY'], ['CREATE TSPARSER', 'CREATE TSTEMPLATE', 'CREATE TYPE', 'CREATE USER', 'CREATE USER MAPPING', 'CREATE VIEW'], ['DEALLOCATE', 'DECLARE', 'DELETE', 'DISCARD', 'DO', 'DROP AGGREGATE'], ['DROP CAST', 'DROP COLLATION', 'DROP CONVERSION', 'DROP DATABASE', 'DROP DOMAIN', 'DROP EVENT TRIGGER'], ['DROP EXTENSION', 'DROP FOREIGN DATA WRAPPER', 'DROP FOREIGN TABLE', 'DROP FUNCTION', 'DROP GROUP', 'DROP INDEX'], ['DROP LANGUAGE', 'DROP MATERIALIZED VIEW', 'DROP OPCLASS', 'DROP OPERATOR', 'DROP OPFAMILY', 'DROP OWNED'], ['DROP POLICY', 'DROP ROLE', 'DROP RULE', 'DROP SCHEMA', 'DROP SEQUENCE', 'DROP SERVER'], ['DROP TABLE', 'DROP TABLESPACE', 'DROP TRANSFORM', 'DROP TRIGGER', 'DROP TSCONFIG', 'DROP TSDICTIONARY'], ['DROP TSPARSER', 'DROP TSTEMPLATE', 'DROP TYPE', 'DROP USER', 'DROP USER MAPPING', 'DROP VIEW'], ['END', 'EXECUTE', 'EXPLAIN', 'FETCH', 'GRANT', 'IMPORT FOREIGN SCHEMA'], ['INSERT', 'LISTEN', 'LOAD', 'LOCK', 'MOVE', 'NOTIFY'], ['PGBENCH', 'PREPARE', 'PREPARE TRANSACTION', 'REASSIGN OWNED', 'REFRESH MATERIALIZED VIEW', 'REINDEX'], ['RELEASE SAVEPOINT', 'RESET', 'REVOKE', 'ROLLBACK', 'ROLLBACK PREPARED', 'ROLLBACK TO'], ['SAVEPOINT', 'SECURITY LABEL', 'SELECT', 'SELECT INTO', 'SET', 'SET CONSTRAINTS'], ['SET ROLE', 'SET SESSION AUTH', 'SET TRANSACTION', 'SHOW', 'START TRANSACTION', 'TRUNCATE'], ['UNLISTEN', 'UPDATE', 'VACUUM', 'VALUES']]

@dbtest
def test_slash_h(executor):
    """List all commands."""
    results = executor('\h')
    expected = [None, help_rows, [], None]
    assert results == expected

@dbtest
def test_slash_h_command(executor):
    """Check help is returned for all commands"""
    for command in itertools.chain(*help_rows):
        results = executor('\h %s' % command)
        assert results[3].startswith('Description\n')
        assert 'Syntax' in results[3]

@dbtest
def test_slash_h_alias(executor):
    """\? is properly aliased to \h"""
    h_results = executor('\h SELECT')
    results = executor('\? SELECT')
    assert results[3] == h_results[3]


@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_stdout(executor, capsys):
    executor(u"\copy (SELECT 'Montréal', 'Portland', 'Cleveland') TO stdout")
    (out, err) = capsys.readouterr()
    assert out == u'Montréal\tPortland\tCleveland\n'


@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_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_sf(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_sf_unknown(executor):
    try:
        executor('\sf non_existing')
    except Exception as e:
        assert 'non_existing' in str(e)
    else:
        assert False, "Expected an exception"


@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_sf_verbose(executor):
    results = executor('\sf+ schema1.s1_func1')
    title = None
    rows = [('        CREATE OR REPLACE FUNCTION schema1.s1_func1()\n'
             '         RETURNS integer\n'
             '         LANGUAGE sql\n'
             '1       AS $function$select 2$function$\n',),
            ]
    headers = ['source']
    status = None
    expected = [title, rows, headers, status]
    assert results == expected

@dbtest
def test_slash_dE(executor):

    with foreign_db_environ():
        results = executor('\dE')
        title = None
        rows = [('public', 'foreign_foo', 'foreign table', 'postgres')]
        headers =  ['Schema', 'Name', 'Type', 'Owner']
        status = 'SELECT 1'
        expected = [title, rows, headers, status]
        assert results == expected

@dbtest
def test_slash_dE_with_pattern(executor):

    with foreign_db_environ():
        results = executor('\dE foreign_foo')
        title = None
        rows = [('public', 'foreign_foo', 'foreign table', 'postgres')]
        headers =  ['Schema', 'Name', 'Type', 'Owner']
        status = 'SELECT 1'
        expected = [title, rows, headers, status]
        assert results == expected

        results = executor('\dE *_foo')
        assert results == expected

        results = executor('\dE no_such_table')
        rows = []
        status = 'SELECT 0'
        expected = [title, rows, headers, status]
        assert results == expected

@dbtest
def test_slash_dE_verbose(executor):

    with foreign_db_environ():
        results = executor('\dE+')
        title = None
        rows = [('public', 'foreign_foo', 'foreign table', 'postgres', '0 bytes', None)]
        headers = ['Schema', 'Name', 'Type', 'Owner', 'Size', 'Description']
        status = 'SELECT 1'
        expected = [title, rows, headers, status]
        assert results == expected