-
Notifications
You must be signed in to change notification settings - Fork 47
/
Copy pathtest_protocol.py
125 lines (100 loc) · 3.57 KB
/
test_protocol.py
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
from django_elasticache.cluster_utils import (
get_cluster_info, WrongProtocolData)
from nose.tools import eq_, raises
import sys
if sys.version < '3':
from mock import patch, call, MagicMock
else:
from unittest.mock import patch, call, MagicMock
TEST_PROTOCOL_1_READ_UNTIL = [
b'VERSION 1.4.14',
]
TEST_PROTOCOL_1_EXPECT = [
(0, None, b'CONFIG cluster 0 138\r\n1\nhost|ip|port host||port\n\r\nEND\r\n'), # NOQA
]
TEST_PROTOCOL_2_READ_UNTIL = [
b'VERSION 1.4.13',
]
TEST_PROTOCOL_2_EXPECT = [
(0, None, b'CONFIG cluster 0 138\r\n1\nhost|ip|port host||port\n\r\nEND\r\n'), # NOQA
]
TEST_PROTOCOL_3_READ_UNTIL = [
b'VERSION 1.4.14 (Ubuntu)',
]
TEST_PROTOCOL_3_EXPECT = [
(0, None, b'CONFIG cluster 0 138\r\n1\nhost|ip|port host||port\n\r\nEND\r\n'), # NOQA
]
TEST_PROTOCOL_4_READ_UNTIL = [
b'VERSION 1.4.34',
]
TEST_PROTOCOL_4_EXPECT = [
(0, None, b'ERROR\r\n'),
]
@patch('django_elasticache.cluster_utils.Telnet')
def test_happy_path(Telnet):
client = Telnet.return_value
client.read_until.side_effect = TEST_PROTOCOL_1_READ_UNTIL
client.expect.side_effect = TEST_PROTOCOL_1_EXPECT
info = get_cluster_info('', 0)
eq_(info['version'], 1)
eq_(info['nodes'], ['ip:port', 'host:port'])
@raises(WrongProtocolData)
@patch('django_elasticache.cluster_utils.Telnet', MagicMock())
def test_bad_protocol():
get_cluster_info('', 0)
@patch('django_elasticache.cluster_utils.Telnet')
def test_last_versions(Telnet):
client = Telnet.return_value
client.read_until.side_effect = TEST_PROTOCOL_1_READ_UNTIL
client.expect.side_effect = TEST_PROTOCOL_1_EXPECT
get_cluster_info('', 0)
client.write.assert_has_calls([
call(b'version\n'),
call(b'config get cluster\n'),
])
@patch('django_elasticache.cluster_utils.Telnet')
def test_prev_versions(Telnet):
client = Telnet.return_value
client.read_until.side_effect = TEST_PROTOCOL_2_READ_UNTIL
client.expect.side_effect = TEST_PROTOCOL_2_EXPECT
get_cluster_info('', 0)
client.write.assert_has_calls([
call(b'version\n'),
call(b'get AmazonElastiCache:cluster\n'),
])
@patch('django_elasticache.cluster_utils.Telnet')
def test_ubuntu_protocol(Telnet):
client = Telnet.return_value
client.read_until.side_effect = TEST_PROTOCOL_3_READ_UNTIL
client.expect.side_effect = TEST_PROTOCOL_3_EXPECT
try:
get_cluster_info('', 0)
except WrongProtocolData:
raise AssertionError('Raised WrongProtocolData with Ubuntu version.')
client.write.assert_has_calls([
call(b'version\n'),
call(b'config get cluster\n'),
])
@patch('django_elasticache.cluster_utils.Telnet')
def test_no_configuration_protocol_support_with_errors_ignored(Telnet):
client = Telnet.return_value
client.read_until.side_effect = TEST_PROTOCOL_4_READ_UNTIL
client.expect.side_effect = TEST_PROTOCOL_4_EXPECT
info = get_cluster_info('test', 0, ignore_cluster_errors=True)
client.write.assert_has_calls([
call(b'version\n'),
call(b'config get cluster\n'),
])
eq_(info['version'], b'1.4.34')
eq_(info['nodes'], ['test:0'])
@raises(WrongProtocolData)
@patch('django_elasticache.cluster_utils.Telnet')
def test_no_configuration_protocol_support_with_errors(Telnet):
client = Telnet.return_value
client.read_until.side_effect = TEST_PROTOCOL_4_READ_UNTIL
client.expect.side_effect = TEST_PROTOCOL_4_EXPECT
get_cluster_info('test', 0, ignore_cluster_errors=False)
client.write.assert_has_calls([
call(b'version\n'),
call(b'config get cluster\n'),
])