forked from twisted/twisted
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_postfix.py
139 lines (118 loc) · 4.34 KB
/
test_postfix.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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Test cases for twisted.protocols.postfix module.
"""
from typing import List, Dict, Tuple
from twisted.trial import unittest
from twisted.protocols import postfix
from twisted.test.proto_helpers import StringTransport
class PostfixTCPMapQuoteTests(unittest.TestCase):
data = [
# (raw, quoted, [aliasQuotedForms]),
(b"foo", b"foo"),
(b"foo bar", b"foo%20bar"),
(b"foo\tbar", b"foo%09bar"),
(b"foo\nbar", b"foo%0Abar", b"foo%0abar"),
(
b"foo\r\nbar",
b"foo%0D%0Abar",
b"foo%0D%0abar",
b"foo%0d%0Abar",
b"foo%0d%0abar",
),
(b"foo ", b"foo%20"),
(b" foo", b"%20foo"),
]
def testData(self):
for entry in self.data:
raw = entry[0]
quoted = entry[1:]
self.assertEqual(postfix.quote(raw), quoted[0])
for q in quoted:
self.assertEqual(postfix.unquote(q), raw)
class PostfixTCPMapServerTestCase:
data = {
# 'key': 'value',
} # type: Dict[bytes, bytes]
chat = [
# (input, expected_output),
] # type: List[Tuple[bytes, bytes]]
def test_chat(self):
"""
Test that I{get} and I{put} commands are responded to correctly by
L{postfix.PostfixTCPMapServer} when its factory is an instance of
L{postifx.PostfixTCPMapDictServerFactory}.
"""
factory = postfix.PostfixTCPMapDictServerFactory(self.data)
transport = StringTransport()
protocol = postfix.PostfixTCPMapServer()
protocol.service = factory
protocol.factory = factory
protocol.makeConnection(transport)
for input, expected_output in self.chat:
protocol.lineReceived(input)
self.assertEqual(
transport.value(),
expected_output,
"For %r, expected %r but got %r"
% (input, expected_output, transport.value()),
)
transport.clear()
protocol.setTimeout(None)
def test_deferredChat(self):
"""
Test that I{get} and I{put} commands are responded to correctly by
L{postfix.PostfixTCPMapServer} when its factory is an instance of
L{postifx.PostfixTCPMapDeferringDictServerFactory}.
"""
factory = postfix.PostfixTCPMapDeferringDictServerFactory(self.data)
transport = StringTransport()
protocol = postfix.PostfixTCPMapServer()
protocol.service = factory
protocol.factory = factory
protocol.makeConnection(transport)
for input, expected_output in self.chat:
protocol.lineReceived(input)
self.assertEqual(
transport.value(),
expected_output,
"For {!r}, expected {!r} but got {!r}".format(
input, expected_output, transport.value()
),
)
transport.clear()
protocol.setTimeout(None)
def test_getException(self):
"""
If the factory throws an exception,
error code 400 must be returned.
"""
class ErrorFactory:
"""
Factory that raises an error on key lookup.
"""
def get(self, key):
raise Exception("This is a test error")
server = postfix.PostfixTCPMapServer()
server.factory = ErrorFactory()
server.transport = StringTransport()
server.lineReceived(b"get example")
self.assertEqual(server.transport.value(), b"400 This is a test error\n")
class ValidTests(PostfixTCPMapServerTestCase, unittest.TestCase):
data = {
b"foo": b"ThisIs Foo",
b"bar": b" bar really is found\r\n",
}
chat = [
(b"get", b"400 Command 'get' takes 1 parameters.\n"),
(b"get foo bar", b"500 \n"),
(b"put", b"400 Command 'put' takes 2 parameters.\n"),
(b"put foo", b"400 Command 'put' takes 2 parameters.\n"),
(b"put foo bar baz", b"500 put is not implemented yet.\n"),
(b"put foo bar", b"500 put is not implemented yet.\n"),
(b"get foo", b"200 ThisIs%20Foo\n"),
(b"get bar", b"200 %20bar%20really%20is%20found%0D%0A\n"),
(b"get baz", b"500 \n"),
(b"foo", b"400 unknown command\n"),
]