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
|
# fmt: off
import pytest
import subprocess
import sys
from typing import List
from conftest import ShellTest
import os
from pathlib import Path
def test_dump_create(shell):
test = (
ShellTest(shell)
.statement("CREATE TABLE a (i INTEGER);")
.statement(".changes off")
.statement("INSERT INTO a VALUES (42);")
.statement(".dump")
)
result = test.run()
result.check_stdout('CREATE TABLE a(i INTEGER)')
result.check_stdout('COMMIT')
@pytest.mark.parametrize("pattern", [
"a",
"a%"
])
def test_dump_specific(shell, pattern):
test = (
ShellTest(shell)
.statement("CREATE TABLE a (i INTEGER);")
.statement(".changes off")
.statement("INSERT INTO a VALUES (42);")
.statement(f".dump {pattern}")
)
result = test.run()
result.check_stdout('CREATE TABLE a(i INTEGER)')
# Original comment: more types, tables and views
def test_dump_mixed(shell):
test = (
ShellTest(shell)
.statement("CREATE TABLE a (d DATE, k FLOAT, t TIMESTAMP);")
.statement("CREATE TABLE b (c INTEGER);")
.statement(".changes off")
.statement("INSERT INTO a VALUES (DATE '1992-01-01', 0.3, NOW());")
.statement("INSERT INTO b SELECT * FROM range(0,10);")
.statement(".dump")
)
result = test.run()
result.check_stdout('CREATE TABLE a(d DATE, k FLOAT, t TIMESTAMP);')
def test_dump_blobs(shell):
test = (
ShellTest(shell)
.statement("create table test(t VARCHAR, b BLOB);")
.statement(".changes off")
.statement("insert into test values('literal blob', '\\x07\\x08\\x09');")
.statement(".dump")
)
result = test.run()
result.check_stdout("'\\x07\\x08\\x09'")
def test_dump_newline(shell):
test = (
ShellTest(shell)
.statement("create table newline_data as select concat(chr(10), '\n') s;")
.statement(".dump")
)
result = test.run()
result.check_stdout("concat")
result.check_stdout("chr(10)")
def test_dump_indexes(shell):
test = (
ShellTest(shell)
.statement("create table integer(i int);")
.statement("create index i_index on integer(i);")
.statement(".dump")
)
result = test.run()
result.check_stdout("CREATE INDEX i_index")
def test_dump_views(shell):
test = (
ShellTest(shell)
.statement("create table integer(i int);")
.statement("create view v1 as select * from integer;")
.statement(".dump")
)
result = test.run()
result.check_stdout("CREATE VIEW v1")
def test_dump_schema_qualified(shell):
test = (
ShellTest(shell)
.statement("CREATE SCHEMA other;")
.statement("CREATE TABLE other.t_in_other(a INT);")
.statement(".dump")
)
result = test.run()
result.check_stdout('CREATE SCHEMA IF NOT EXISTS other;')
result.check_stdout('CREATE TABLE other.t_in_other(a INTEGER);')
result.check_stdout('COMMIT')
def test_dump_schema_with_data(shell):
test = (
ShellTest(shell)
.statement("CREATE SCHEMA test_schema;")
.statement("CREATE TABLE test_schema.tbl(x INT, y VARCHAR);")
.statement(".changes off")
.statement("INSERT INTO test_schema.tbl VALUES (1, 'hello'), (2, 'world');")
.statement(".dump")
)
result = test.run()
result.check_stdout('CREATE SCHEMA IF NOT EXISTS test_schema;')
result.check_stdout('CREATE TABLE test_schema.tbl(x INTEGER, y VARCHAR);')
result.check_stdout("INSERT INTO test_schema.tbl VALUES(1,'hello');")
result.check_stdout('COMMIT')
def test_dump_multiple_schemas(shell):
test = (
ShellTest(shell)
.statement("CREATE SCHEMA s1;")
.statement("CREATE SCHEMA s2;")
.statement("CREATE TABLE s1.t1(a INT);")
.statement("CREATE TABLE s2.t2(b INT);")
.statement(".changes off")
.statement("INSERT INTO s1.t1 VALUES (10);")
.statement("INSERT INTO s2.t2 VALUES (20);")
.statement(".dump")
)
result = test.run()
result.check_stdout('CREATE SCHEMA IF NOT EXISTS s1;')
result.check_stdout('CREATE SCHEMA IF NOT EXISTS s2;')
result.check_stdout('INSERT INTO s1.t1 VALUES(10);')
result.check_stdout('INSERT INTO s2.t2 VALUES(20);')
def test_dump_quoted_schema(shell):
test = (
ShellTest(shell)
.statement('CREATE SCHEMA "my-schema";')
.statement('CREATE TABLE "my-schema"."my-table"(a INT);')
.statement(".dump")
)
result = test.run()
result.check_stdout('CREATE SCHEMA IF NOT EXISTS "my-schema";')
result.check_stdout('CREATE TABLE "my-schema"."my-table"(a INTEGER);')
def test_dump_if_not_exists(shell):
test = (
ShellTest(shell)
.statement("CREATE SCHEMA other;")
.statement("CREATE TABLE IF NOT EXISTS other.tbl(x INT);")
.statement(".changes off")
.statement("INSERT INTO other.tbl VALUES (42);")
.statement(".dump")
)
result = test.run()
result.check_stdout('CREATE SCHEMA IF NOT EXISTS other;')
result.check_stdout('INSERT INTO other.tbl VALUES(42);')
result.check_stdout('COMMIT')
|
