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
|
# type: ignore
"""Steps for behavioral style tests are defined in this module.
Each step is defined by the string decorating it. This string is used
to call the step in "*.feature" file.
"""
from textwrap import dedent
from behave import then, when
import wrappers
@when("we create table")
def step_create_table(context):
"""Send create table."""
context.cli.sendline("create table a(x text);")
@when("we insert into table")
def step_insert_into_table(context):
"""Send insert into table."""
context.cli.sendline("""insert into a(x) values('xxx');""")
@when("we update table")
def step_update_table(context):
"""Send insert into table."""
context.cli.sendline("""update a set x = 'yyy' where x = 'xxx';""")
@when("we select from table")
def step_select_from_table(context):
"""Send select from table."""
context.cli.sendline("select * from a;")
@when("we delete from table")
def step_delete_from_table(context):
"""Send deete from table."""
context.cli.sendline("""delete from a where x = 'yyy';""")
@when("we drop table")
def step_drop_table(context):
"""Send drop table."""
context.cli.sendline("drop table a;")
@then("we see table created")
def step_see_table_created(context):
"""Wait to see create table output."""
wrappers.expect_exact(context, "Query OK, 0 rows affected", timeout=2)
@then("we see record inserted")
def step_see_record_inserted(context):
"""Wait to see insert output."""
wrappers.expect_exact(context, "Query OK, 1 row affected", timeout=2)
@then("we see record updated")
def step_see_record_updated(context):
"""Wait to see update output."""
wrappers.expect_exact(context, "Query OK, 1 row affected", timeout=2)
@then("we see data selected")
def step_see_data_selected(context):
"""Wait to see select output."""
expected = (
dedent(
"""
+-----+\r
| x |\r
+-----+\r
| yyy |\r
+-----+
"""
).strip()
+ '\r\n\r\n'
)
wrappers.expect_pager(
context,
expected,
timeout=2,
)
wrappers.expect_exact(context, "1 row in set", timeout=2)
@then("we see record deleted")
def step_see_data_deleted(context):
"""Wait to see delete output."""
wrappers.expect_exact(context, "Query OK, 1 row affected", timeout=2)
@then("we see table dropped")
def step_see_table_dropped(context):
"""Wait to see drop output."""
wrappers.expect_exact(context, "Query OK, 0 rows affected", timeout=2)
@when("we select null")
def step_select_null(context):
"""Send select null."""
context.cli.sendline("select null;")
@then("we see null selected")
def step_see_null_selected(context):
"""Wait to see null output."""
expected = (
dedent(
"""
+--------+\r
| NULL |\r
+--------+\r
| <null> |\r
+--------+
"""
).strip()
+ '\r\n\r\n'
)
wrappers.expect_pager(
context,
expected,
timeout=2,
)
wrappers.expect_exact(context, "1 row in set", timeout=2)
|
