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
|
# -*- coding: utf-8 -*-
"""
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 __future__ import unicode_literals, print_function
import pexpect
from behave import when, then
import wrappers
@when('we create database')
def step_db_create(context):
"""
Send create database.
"""
context.cli.sendline('create database {0};'.format(
context.conf['dbname_tmp']))
context.response = {
'database_name': context.conf['dbname_tmp']
}
@when('we drop database')
def step_db_drop(context):
"""
Send drop database.
"""
context.cli.sendline('drop database {0};'.format(
context.conf['dbname_tmp']))
@when('we connect to test database')
def step_db_connect_test(context):
"""
Send connect to database.
"""
db_name = context.conf['dbname']
context.cli.sendline('\\connect {0}'.format(db_name))
@when('we connect to dbserver')
def step_db_connect_dbserver(context):
"""
Send connect to database.
"""
context.cli.sendline('\\connect postgres')
context.currentdb = 'postgres'
@then('dbcli exits')
def step_wait_exit(context):
"""
Make sure the cli exits.
"""
wrappers.expect_exact(context, pexpect.EOF, timeout=5)
@then('we see dbcli prompt')
def step_see_prompt(context):
"""
Wait to see the prompt.
"""
wrappers.expect_exact(context, '{0}> '.format(context.conf['dbname']), timeout=5)
context.atprompt = True
@then('we see help output')
def step_see_help(context):
for expected_line in context.fixture_data['help_commands.txt']:
wrappers.expect_exact(context, expected_line, timeout=2)
@then('we see database created')
def step_see_db_created(context):
"""
Wait to see create database output.
"""
wrappers.expect_pager(context, 'CREATE DATABASE\r\n', timeout=5)
@then('we see database dropped')
def step_see_db_dropped(context):
"""
Wait to see drop database output.
"""
wrappers.expect_pager(context, 'DROP DATABASE\r\n', timeout=2)
@then('we see database connected')
def step_see_db_connected(context):
"""
Wait to see drop database output.
"""
wrappers.expect_exact(context, 'You are now connected to database', timeout=2)
|
