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
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
# -*- coding: utf-8 -*-
"""
License: BSD
(c) 2009 ::: www.CodeResort.com - BV Network AS (simon-code@bvnetwork.no)
"""
import sys
import unittest
from datetime import datetime
from trac.util.datefmt import to_datetime, utc
from ..util import xmlrpclib
from ..xml_rpc import (to_xmlrpc_datetime, from_xmlrpc_datetime,
_illegal_unichrs, REPLACEMENT_CHAR)
from . import (Request, TracRpcTestCase, TracRpcTestSuite, b64encode, urlopen,
makeSuite)
class RpcXmlTestCase(TracRpcTestCase):
def setUp(self):
TracRpcTestCase.setUp(self)
self.anon = xmlrpclib.ServerProxy(self._testenv.url_anon)
self.user = xmlrpclib.ServerProxy(self._testenv.url_user)
self.admin = xmlrpclib.ServerProxy(self._testenv.url_admin)
def tearDown(self):
for proxy in (self.anon, self.user, self.admin):
proxy('close')()
TracRpcTestCase.tearDown(self)
def test_xmlrpc_permission(self):
# Test returned response if not XML_RPC permission
self._revoke_perm('anonymous', 'XML_RPC')
try:
self.anon.system.listMethods()
except xmlrpclib.Fault as e:
self.assertEqual(403, e.faultCode)
self.assertIn('XML_RPC', e.faultString)
else:
self.fail('xmlrpclib.Fault not raised')
finally:
self._grant_perm('anonymous', 'XML_RPC')
def test_method_not_found(self):
try:
self.admin.system.doesNotExist()
except xmlrpclib.Fault as e:
self.assertEqual(-32601, e.faultCode)
self.assertIn("not found", e.faultString)
else:
self.fail('xmlrpclib.Fault not raised')
def test_wrong_argspec(self):
try:
self.admin.system.listMethods("hello")
except xmlrpclib.Fault as e:
self.assertEqual(1, e.faultCode)
if sys.version_info[0] == 2:
self.assertIn('listMethods() takes exactly 2 arguments',
e.faultString)
else:
self.assertIn('listMethods() takes 2 positional arguments but '
'3 were given', e.faultString)
else:
self.fail('xmlrpclib.Fault not raised')
def test_content_encoding(self):
test_bytes = u'øæåØÆÅàéüoö'.encode('utf-8')
body = (b'<?xml version="1.0"?>\n'
b'<methodCall>\n'
b' <methodName>ticket.create</methodName>\n'
b' <params>\n'
b' <param><string>%s</string></param>\n'
b' <param><string>%s</string></param>\n'
b' </params>\n'
b'</methodCall>' % (test_bytes, test_bytes[::-1]))
request = Request(self._testenv.url_auth, data=body)
request.add_header('Content-Type', 'application/xml')
request.add_header('Content-Length', str(len(body)))
request.add_header('Authorization',
'Basic %s' % b64encode('admin:admin'))
self.assertEqual('POST', request.get_method())
response = urlopen(request)
self.assertEqual(200, response.code)
self.assertIn(b'<member>\n'
b'<name>faultCode</name>\n'
b'<value><int>-32700</int></value>\n'
b'</member>', response.read())
def test_to_and_from_datetime(self):
now = to_datetime(None, utc)
now_timetuple = now.timetuple()[:6]
xmlrpc_now = to_xmlrpc_datetime(now)
self.assertTrue(isinstance(xmlrpc_now, xmlrpclib.DateTime),
"Expected xmlprc_now to be an xmlrpclib.DateTime")
self.assertEqual(str(xmlrpc_now), now.strftime("%Y%m%dT%H:%M:%S"))
now_from_xmlrpc = from_xmlrpc_datetime(xmlrpc_now)
self.assertTrue(isinstance(now_from_xmlrpc, datetime),
"Expected now_from_xmlrpc to be a datetime")
self.assertEqual(now_from_xmlrpc.timetuple()[:6], now_timetuple)
self.assertEqual(now_from_xmlrpc.tzinfo, utc)
def test_resource_not_found(self):
# A Ticket resource
try:
self.admin.ticket.get(2147483647)
except xmlrpclib.Fault as e:
self.assertEqual(e.faultCode, 404)
self.assertEqual(e.faultString,
'Ticket 2147483647 does not exist.')
else:
self.fail('xmlrpclib.Fault not raised')
# A Wiki resource
try:
self.admin.wiki.getPage("Test", 10)
except xmlrpclib.Fault as e:
self.assertEqual(e.faultCode, 404)
self.assertEqual(e.faultString,
'Wiki page "Test" does not exist at version 10')
else:
self.fail('xmlrpclib.Fault not raised')
@unittest.expectedFailure
def test_xml_encoding_special_characters(self):
tid1 = self.admin.ticket.create(
'One & Two < Four', 'Desc & ription\nLine 2', {})
ticket = self.admin.ticket.get(tid1)
try:
self.assertEqual('One & Two < Four', ticket[3]['summary'])
self.assertEqual('Desc & ription\r\nLine 2',
ticket[3]['description'])
finally:
self.admin.ticket.delete(tid1)
def test_xml_encoding_invalid_characters(self):
# Enable ticket manipulator
source = r"""# -*- coding: utf-8 -*-
from trac.core import *
from tracrpc.api import IXMLRPCHandler
class UniChr(Component):
implements(IXMLRPCHandler)
def xmlrpc_namespace(self):
return 'test_unichr'
def xmlrpc_methods(self):
yield ('XML_RPC', ((str, int),), self.unichr)
def unichr(self, req, code):
return (b'\\U%08X' % code).decode('unicode-escape')
"""
with self._plugin(source, 'InvalidXmlCharHandler.py'):
for low, high in _illegal_unichrs:
for code in sorted(set([low, low + 1, high - 1, high])):
self.assertEqual(REPLACEMENT_CHAR,
self.user.test_unichr.unichr(code),
"Failed unichr with U+%04X" % code)
# surrogate pair on narrow build
self.assertEqual(u'\U0001D4C1',
self.user.test_unichr.unichr(0x1D4C1))
def test_large_file(self):
pagename = 'SandBox/LargeXmlrpc'
filename = 'large.dat'
rv = self.admin.wiki.putPage(pagename, 'attachment:' + filename, {})
self.assertEqual(True, rv)
content = bytes(bytearray(range(256))) * 4 * 1024 * 4 # 4 MB
rv = self.admin.wiki.putAttachmentEx(pagename, filename, 'Large file',
xmlrpclib.Binary(content))
self.assertEqual(filename, rv)
rv = self.admin.wiki.getAttachment('%s/%s' % (pagename, filename))
self.assertIsInstance(rv, xmlrpclib.Binary)
self.assertEqual(content, rv.data)
def test_suite():
suite = TracRpcTestSuite()
suite.addTest(makeSuite(RpcXmlTestCase))
return suite
if __name__ == '__main__':
unittest.main(defaultTest='test_suite')
|
